Merge perch.ndb.mysql.com:/home/jonas/src/51-jonas

into  perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new-ndb

mysql-test/extra/rpl_tests/rpl_row_blob.test

@@ -183,3 +183,9 @@ connection master;
 
 DROP TABLE IF EXISTS test.t1;
 DROP TABLE IF EXISTS test.t2;
+# ensure cleanup on slave as well:
+# ndb blob tables consist of several tables
+# if cluster is shutdown while not all tables are
+# properly dropped, the table becomes inconsistent
+# and wrecks later test cases
+--sync_slave_with_master

sql/ha_ndbcluster.cc

@@ -4480,6 +4480,21 @@ int ha_ndbcluster::create(const char *name,
     DBUG_RETURN(my_errno);
   }
 
+#ifdef HAVE_NDB_BINLOG
+  /*
+    Don't allow table creation unless
+    schema distribution table is setup
+    ( unless it is a creation of the schema dist table itself )
+  */
+  if (!schema_share &&
+      !(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);
+  }
+#endif /* HAVE_NDB_BINLOG */
+
   DBUG_PRINT("table", ("name: %s", m_tabname));  
   tab.setName(m_tabname);
   tab.setLogging(!(info->options & HA_LEX_CREATE_TMP_TABLE));    
@@ -5004,7 +5019,8 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
     is_old_table_tmpfile= 0;
     String event_name(INJECTOR_EVENT_LEN);
     ndb_rep_event_name(&event_name, from + sizeof(share_prefix) - 1, 0);
-    ndbcluster_handle_drop_table(ndb, event_name.c_ptr(), share);
+    ndbcluster_handle_drop_table(ndb, event_name.c_ptr(), share,
+                                 "rename table");
   }
 
   if (!result && !IS_TMP_PREFIX(new_tabname))
@@ -5088,6 +5104,15 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb,
   DBUG_ENTER("ha_ndbcluster::ndbcluster_delete_table");
   NDBDICT *dict= ndb->getDictionary();
 #ifdef HAVE_NDB_BINLOG
+  /*
+    Don't allow drop table unless
+    schema distribution table is setup
+  */
+  if (!schema_share)
+  {
+    DBUG_PRINT("info", ("Schema distribution table not setup"));
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);
+  }
   NDB_SHARE *share= get_share(path, 0, false);
 #endif
 
@@ -5156,7 +5181,7 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb,
     ndb_rep_event_name(&event_name, path + sizeof(share_prefix) - 1, 0);
     ndbcluster_handle_drop_table(ndb,
                                  table_dropped ? event_name.c_ptr() : 0,
-                                 share);
+                                 share, "delete table");
   }
 
   if (share)
@@ -5185,6 +5210,18 @@ int ha_ndbcluster::delete_table(const char *name)
   set_dbname(name);
   set_tabname(name);
 
+#ifdef HAVE_NDB_BINLOG
+  /*
+    Don't allow drop table unless
+    schema distribution table is setup
+  */
+  if (!schema_share)
+  {
+    DBUG_PRINT("info", ("Schema distribution table not setup"));
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);
+  }
+#endif
+
   if (check_ndb_connection())
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
 
@@ -5406,6 +5443,11 @@ int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
   if (!res)
     info(HA_STATUS_VARIABLE | HA_STATUS_CONST);
 
+#ifdef HAVE_NDB_BINLOG
+  if (!ndb_binlog_tables_inited && ndb_binlog_running)
+    table->db_stat|= HA_READ_ONLY;
+#endif
+
   DBUG_RETURN(res);
 }
 
@@ -5704,6 +5746,19 @@ int ndbcluster_drop_database_impl(const char *path)
 
 static void ndbcluster_drop_database(char *path)
 {
+  DBUG_ENTER("ndbcluster_drop_database");
+#ifdef HAVE_NDB_BINLOG
+  /*
+    Don't allow drop database unless
+    schema distribution table is setup
+  */
+  if (!schema_share)
+  {
+    DBUG_PRINT("info", ("Schema distribution table not setup"));
+    DBUG_VOID_RETURN;
+    //DBUG_RETURN(HA_ERR_NO_CONNECTION);
+  }
+#endif
   ndbcluster_drop_database_impl(path);
 #ifdef HAVE_NDB_BINLOG
   char db[FN_REFLEN];
@@ -5712,6 +5767,7 @@ static void ndbcluster_drop_database(char *path)
                            current_thd->query, current_thd->query_length,
                            db, "", 0, 0, SOT_DROP_DB);
 #endif
+  DBUG_VOID_RETURN;
 }
 /*
   find all tables in ndb and discover those needed
@@ -5733,36 +5789,37 @@ int ndbcluster_find_all_files(THD *thd)
   DBUG_ENTER("ndbcluster_find_all_files");
   Ndb* ndb;
   char key[FN_REFLEN];
-  NdbDictionary::Dictionary::List list;
 
   if (!(ndb= check_ndb_in_thd(thd)))
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
 
   NDBDICT *dict= ndb->getDictionary();
 
-  int unhandled, retries= 5;
+  int unhandled, retries= 5, skipped;
   do
   {
+    NdbDictionary::Dictionary::List list;
     if (dict->listObjects(list, NdbDictionary::Object::UserTable) != 0)
       ERR_RETURN(dict->getNdbError());
     unhandled= 0;
+    skipped= 0;
+    retries--;
     for (uint i= 0 ; i < list.count ; i++)
     {
       NDBDICT::List::Element& elmt= list.elements[i];
-      int do_handle_table= 0;
       if (IS_TMP_PREFIX(elmt.name) || IS_NDB_BLOB_PREFIX(elmt.name))
       {
         DBUG_PRINT("info", ("Skipping %s.%s in NDB", elmt.database, elmt.name));
         continue;
       }
       DBUG_PRINT("info", ("Found %s.%s in NDB", elmt.database, elmt.name));
-      if (elmt.state == NDBOBJ::StateOnline ||
-          elmt.state == NDBOBJ::StateBackup)
-        do_handle_table= 1;
-      else if (!(elmt.state == NDBOBJ::StateBuilding))
+      if (elmt.state != NDBOBJ::StateOnline &&
+          elmt.state != NDBOBJ::StateBackup &&
+          elmt.state != NDBOBJ::StateBuilding)
       {
         sql_print_information("NDB: skipping setup table %s.%s, in state %d",
                               elmt.database, elmt.name, elmt.state);
+        skipped++;
         continue;
       }
 
@@ -5771,7 +5828,7 @@ int ndbcluster_find_all_files(THD *thd)
 
       if (!(ndbtab= dict->getTable(elmt.name)))
       {
-        if (do_handle_table)
+        if (retries == 0)
           sql_print_error("NDB: failed to setup table %s.%s, error: %d, %s",
                           elmt.database, elmt.name,
                           dict->getNdbError().code,
@@ -5840,9 +5897,9 @@ int ndbcluster_find_all_files(THD *thd)
       pthread_mutex_unlock(&LOCK_open);
     }
   }
-  while (unhandled && retries--);
+  while (unhandled && retries);
 
-  DBUG_RETURN(0);
+  DBUG_RETURN(-(skipped + unhandled));
 }
 
 int ndbcluster_find_files(THD *thd,const char *db,const char *path,
@@ -7706,6 +7763,8 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
     pthread_cond_wait(&COND_server_started, &LOCK_server_started);
   pthread_mutex_unlock(&LOCK_server_started);
 
+  ndbcluster_util_inited= 1;
+
   /*
     Wait for cluster to start
   */
@@ -7737,6 +7796,8 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
   }
 
 #ifdef HAVE_NDB_BINLOG
+  if (ndb_extra_logging && ndb_binlog_running)
+    sql_print_information("NDB Binlog: Ndb tables initially read only.");
   /* create tables needed by the replication */
   ndbcluster_setup_binlog_table_shares(thd);
 #else
@@ -7746,17 +7807,9 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
   ndbcluster_find_all_files(thd);
 #endif
 
-  ndbcluster_util_inited= 1;
-
-#ifdef HAVE_NDB_BINLOG
-  /* Signal injector thread that all is setup */
-  pthread_cond_signal(&injector_cond);
-#endif
-
   set_timespec(abstime, 0);
   for (;!abort_loop;)
   {
-
     pthread_mutex_lock(&LOCK_ndb_util_thread);
     pthread_cond_timedwait(&COND_ndb_util_thread,
                            &LOCK_ndb_util_thread,
@@ -7774,7 +7827,7 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
       Check that the apply_status_share and schema_share has been created.
       If not try to create it
     */
-    if (!apply_status_share || !schema_share)
+    if (!ndb_binlog_tables_inited)
       ndbcluster_setup_binlog_table_shares(thd);
 #endif
 
@@ -10029,14 +10082,15 @@ static int ndbcluster_fill_files_table(THD *thd, TABLE_LIST *tables, COND *cond)
     }
   }
 
-  dict->listObjects(dflist, NdbDictionary::Object::Undofile);
+  NdbDictionary::Dictionary::List uflist;
+  dict->listObjects(uflist, NdbDictionary::Object::Undofile);
   ndberr= dict->getNdbError();
   if (ndberr.classification != NdbError::NoError)
     ERR_RETURN(ndberr);
 
-  for (i= 0; i < dflist.count; i++)
+  for (i= 0; i < uflist.count; i++)
   {
-    NdbDictionary::Dictionary::List::Element& elt= dflist.elements[i];
+    NdbDictionary::Dictionary::List::Element& elt= uflist.elements[i];
     Ndb_cluster_connection_node_iter iter;
     unsigned id;
 

sql/ha_ndbcluster_binlog.h

@@ -101,7 +101,8 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
                              const char *old_db= 0,
                              const char *old_table_name= 0);
 int ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name,
-                                 NDB_SHARE *share);
+                                 NDB_SHARE *share,
+                                 const char *type_str);
 void ndb_rep_event_name(String *event_name,
                         const char *db, const char *tbl);
 int ndb_create_table_from_engine(THD *thd, const char *db,
@@ -112,12 +113,13 @@ pthread_handler_t ndb_binlog_thread_func(void *arg);
 /*
   table cluster_replication.apply_status
 */
-void ndbcluster_setup_binlog_table_shares(THD *thd);
+int ndbcluster_setup_binlog_table_shares(THD *thd);
 extern NDB_SHARE *apply_status_share;
 extern NDB_SHARE *schema_share;
 
 extern THD *injector_thd;
 extern my_bool ndb_binlog_running;
+extern my_bool ndb_binlog_tables_inited;
 
 bool
 ndbcluster_show_status_binlog(THD* thd, stat_print_fn *stat_print,

storage/ndb/include/kernel/signaldata/DictTabInfo.hpp

@@ -586,7 +586,8 @@ public:
   enum Unimplemented
   {
     ScanOptimised      = 15, //Default updateOptimised
-    AttributeGroup     = 1012 //Default 0
+    AttributeGroup     = 1012, //Default 0
+    FileNo             = 102
   };
 };
 
@@ -618,13 +619,13 @@ struct DictFilegroupInfo {
      */
     FileName          = 100,
     FileType          = 101,
-    FileId            = 102,
-    FileNo            = 103, // Per Filegroup
+    FileId            = 103,
     FileFGroupId      = 104,
     FileFGroupVersion = 105,
     FileSizeHi        = 106,
     FileSizeLo        = 107,
     FileFreeExtents   = 108,
+    FileVersion       = 109,
     FileEnd           = 199, //    
 
     /**
@@ -696,8 +697,8 @@ struct DictFilegroupInfo {
   struct File {
     char FileName[PATH_MAX];
     Uint32 FileType;
-    Uint32 FileNo;
     Uint32 FileId;
+    Uint32 FileVersion;
     Uint32 FilegroupId;
     Uint32 FilegroupVersion;
     Uint32 FileSizeHi;

storage/ndb/include/kernel/signaldata/DropFilegroup.hpp

@@ -66,7 +66,8 @@ struct DropFilegroupRef {
     Busy = 701,
     NotMaster = 702,
     NoSuchFilegroup = 767,
-    FilegroupInUse = 768
+    FilegroupInUse = 768,
+    InvalidSchemaObjectVersion = 774
   };
   
   Uint32 senderData;
@@ -150,7 +151,8 @@ struct DropFileRef {
     NoError = 0,
     Busy = 701,
     NoSuchFile = 766,
-    DropUndoFileNotSupported = 769
+    DropUndoFileNotSupported = 769,
+    InvalidSchemaObjectVersion = 774
   };
 
   Uint32 senderData;

storage/ndb/src/common/debugger/signaldata/DictTabInfo.cpp

@@ -216,8 +216,8 @@ SimpleProperties::SP2StructMapping
 DictFilegroupInfo::FileMapping[] = {
   DFGIMAPS(File, FileName, FileName, 0, PATH_MAX),
   DFGIMAP2(File, FileType, FileType, 0, 1),
-  DFGIMAP(File, FileNo, FileNo),
   DFGIMAP(File, FileId, FileId),
+  DFGIMAP(File, FileVersion, FileVersion),
   DFGIMAP(File, FileFGroupId, FilegroupId),
   DFGIMAP(File, FileFGroupVersion, FilegroupVersion),
   DFGIMAP(File, FileSizeHi, FileSizeHi),
@@ -254,8 +254,8 @@ void
 DictFilegroupInfo::File::init(){
   memset(FileName, sizeof(FileName), 0);
   FileType = ~0;
-  FileNo = ~0;
   FileId = ~0;
+  FileVersion = ~0;
   FilegroupId = ~0;
   FilegroupVersion = ~0;
   FileSizeHi = 0;

storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp

@@ -631,7 +631,8 @@ Dbdict::packFileIntoPages(SimpleProperties::Writer & w,
   f.FileSizeHi = (f_ptr.p->m_file_size >> 32);
   f.FileSizeLo = (f_ptr.p->m_file_size & 0xFFFFFFFF);
   f.FileFreeExtents= free_extents;
-  f.FileNo =  f_ptr.p->key;
+  f.FileId =  f_ptr.p->key;
+  f.FileVersion = f_ptr.p->m_version;
 
   FilegroupPtr lfg_ptr;
   ndbrequire(c_filegroup_hash.find(lfg_ptr, f.FilegroupId));
@@ -13588,6 +13589,13 @@ Dbdict::execDROP_FILE_REQ(Signal* signal)
       break;
     }
     
+    if (file_ptr.p->m_version != version)
+    {
+      ref->errorCode = DropFileRef::InvalidSchemaObjectVersion;
+      ref->errorLine = __LINE__;
+      break;
+    }
+
     Ptr<SchemaTransaction> trans_ptr;
     if (! c_Trans.seize(trans_ptr))
     {
@@ -13663,6 +13671,13 @@ Dbdict::execDROP_FILEGROUP_REQ(Signal* signal)
       break;
     }
 
+    if (filegroup_ptr.p->m_version != version)
+    {
+      ref->errorCode = DropFilegroupRef::InvalidSchemaObjectVersion;
+      ref->errorLine = __LINE__;
+      break;
+    }
+    
     Ptr<SchemaTransaction> trans_ptr;
     if (! c_Trans.seize(trans_ptr))
     {
@@ -15095,6 +15110,7 @@ Dbdict::create_file_prepare_start(Signal* signal, SchemaOp* op){
     filePtr.p->m_obj_ptr_i = obj_ptr.i;
     filePtr.p->m_filegroup_id = f.FilegroupId;
     filePtr.p->m_type = f.FileType;
+    filePtr.p->m_version = op->m_obj_version;
     
     obj_ptr.p->m_id = op->m_obj_id;
     obj_ptr.p->m_type = f.FileType;

storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp

@@ -535,6 +535,7 @@ public:
     
     Uint32 key;
     Uint32 m_magic;
+    Uint32 m_version;
     Uint32 m_obj_ptr_i;
     Uint32 m_filegroup_id;
     Uint32 m_type;

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -4625,12 +4625,26 @@ NdbDictInterface::get_filegroup(NdbFilegroupImpl & dst,
   ptr[0].p  = (Uint32*)name;
   ptr[0].sz = (strLen + 3)/4;
   
+#ifndef IGNORE_VALGRIND_WARNINGS
+  if (strLen & 3)
+  {
+    Uint32 pad = 0;
+    m_buffer.clear();
+    m_buffer.append(name, strLen);
+    m_buffer.append(&pad, 4);
+    ptr[0].p = (Uint32*)m_buffer.get_data();
+  }
+#endif
+  
   int r = dictSignal(&tSignal, ptr, 1,
 		     -1, // any node
 		     WAIT_GET_TAB_INFO_REQ,
 		     DICT_WAITFOR_TIMEOUT, 100);
   if (r)
   {
+    dst.m_id = -1;
+    dst.m_version = ~0;
+    
     DBUG_PRINT("info", ("get_filegroup failed dictSignal"));
     DBUG_RETURN(-1);
   }
@@ -4767,6 +4781,17 @@ NdbDictInterface::get_file(NdbFileImpl & dst,
   ptr[0].p  = (Uint32*)name;
   ptr[0].sz = (strLen + 3)/4;
   
+#ifndef IGNORE_VALGRIND_WARNINGS
+  if (strLen & 3)
+  {
+    Uint32 pad = 0;
+    m_buffer.clear();
+    m_buffer.append(name, strLen);
+    m_buffer.append(&pad, 4);
+    ptr[0].p = (Uint32*)m_buffer.get_data();
+  }
+#endif
+  
   int r = dictSignal(&tSignal, ptr, 1,
 		     node,
 		     WAIT_GET_TAB_INFO_REQ,
@@ -4834,7 +4859,8 @@ NdbDictInterface::parseFileInfo(NdbFileImpl &dst,
   }
 
   dst.m_type= (NdbDictionary::Object::Type)f.FileType;
-  dst.m_id= f.FileNo;
+  dst.m_id= f.FileId;
+  dst.m_version = f.FileVersion;
 
   dst.m_size= ((Uint64)f.FileSizeHi << 32) | (f.FileSizeLo);
   dst.m_path.assign(f.FileName);

storage/ndb/src/ndbapi/ndberror.c

@@ -386,6 +386,7 @@ ErrorBundle ErrorCodes[] = {
   { 768,  DMEC, SE, "Cant drop filegroup, filegroup is used" },
   { 769,  DMEC, SE, "Drop undofile not supported, drop logfile group instead" },
   { 770,  DMEC, SE, "Cant drop file, file is used" },
+  { 774,  DMEC, SE, "Invalid schema object for drop" },
   { 241,  HA_ERR_TABLE_DEF_CHANGED, SE, "Invalid schema object version" },
   { 283,  HA_ERR_NO_SUCH_TABLE, SE, "Table is being dropped" },
   { 284,  HA_ERR_TABLE_DEF_CHANGED, SE, "Table not defined in transaction coordinator" },