Merge joreland@bk-internal.mysql.com:/home/bk/mysql-5.1-new-ndb

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

4 files changed, 86 insertions, 15 deletions

diff --git a/storage/ndb/include/kernel/signaldata/DropFilegroup.hpp b/storage/ndb/include/kernel/signaldata/DropFilegroup.hpp
index 7cf275b1f9e..c1dbc95380c 100644
--- a/storage/ndb/include/kernel/signaldata/DropFilegroup.hpp
+++ b/storage/ndb/include/kernel/signaldata/DropFilegroup.hpp
@@ -150,6 +150,7 @@ struct DropFileRef {
   enum ErrorCode {
     NoError = 0,
     Busy = 701,
+    NotMaster = 702,
     NoSuchFile = 766,
     DropUndoFileNotSupported = 769,
     InvalidSchemaObjectVersion = 774
diff --git a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
index 87bd1d7c53b..57aa9890f24 100644
--- a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
+++ b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
@@ -13396,6 +13396,24 @@ Dbdict::execCREATE_FILE_REQ(Signal* signal){
   Uint32 requestInfo = req->requestInfo;
   
   do {
+    if(getOwnNodeId() != c_masterNodeId){
+      jam();
+      ref->errorCode = CreateFileRef::NotMaster;
+      ref->status    = 0;
+      ref->errorKey  = 0;
+      ref->errorLine = __LINE__;
+      break;
+    }
+    
+    if (c_blockState != BS_IDLE){
+      jam();
+      ref->errorCode = CreateFileRef::Busy;
+      ref->status    = 0;
+      ref->errorKey  = 0;
+      ref->errorLine = __LINE__;
+      break;
+    }
+
     Ptr<SchemaTransaction> trans_ptr;
     if (! c_Trans.seize(trans_ptr)){
       ref->errorCode = CreateFileRef::Busy;
@@ -13455,6 +13473,9 @@ Dbdict::execCREATE_FILE_REQ(Signal* signal){
     tmp.init<CreateObjRef>(rg, GSN_CREATE_OBJ_REF, trans_key);
     sendSignal(rg, GSN_CREATE_OBJ_REQ, signal, 
 	       CreateObjReq::SignalLength, JBB);
+
+    c_blockState = BS_CREATE_TAB;
+
     return;
   } while(0);
   
@@ -13480,15 +13501,6 @@ Dbdict::execCREATE_FILEGROUP_REQ(Signal* signal){
   Uint32 type = req->objType;
   
   do {
-    Ptr<SchemaTransaction> trans_ptr;
-    if (! c_Trans.seize(trans_ptr)){
-      ref->errorCode = CreateFilegroupRef::Busy;
-      ref->status    = 0;
-      ref->errorKey  = 0;
-      ref->errorLine = __LINE__;
-      break;
-    }
-
     if(getOwnNodeId() != c_masterNodeId){
       jam();
       ref->errorCode = CreateFilegroupRef::NotMaster;
@@ -13506,6 +13518,15 @@ Dbdict::execCREATE_FILEGROUP_REQ(Signal* signal){
       ref->errorLine = __LINE__;
       break;
     }
+
+    Ptr<SchemaTransaction> trans_ptr;
+    if (! c_Trans.seize(trans_ptr)){
+      ref->errorCode = CreateFilegroupRef::Busy;
+      ref->status    = 0;
+      ref->errorKey  = 0;
+      ref->errorLine = __LINE__;
+      break;
+    }
     
     const Uint32 trans_key = ++c_opRecordSequence;
     trans_ptr.p->key = trans_key;
@@ -13554,6 +13575,9 @@ Dbdict::execCREATE_FILEGROUP_REQ(Signal* signal){
     tmp.init<CreateObjRef>(rg, GSN_CREATE_OBJ_REF, trans_key);
     sendSignal(rg, GSN_CREATE_OBJ_REQ, signal, 
 	       CreateObjReq::SignalLength, JBB);
+
+    c_blockState = BS_CREATE_TAB;
+
     return;
   } while(0);
   
@@ -13581,6 +13605,22 @@ Dbdict::execDROP_FILE_REQ(Signal* signal)
   Uint32 version = req->file_version;
   
   do {
+    if(getOwnNodeId() != c_masterNodeId){
+      jam();
+      ref->errorCode = DropFileRef::NotMaster;
+      ref->errorKey  = 0;
+      ref->errorLine = __LINE__;
+      break;
+    }
+    
+    if (c_blockState != BS_IDLE){
+      jam();
+      ref->errorCode = DropFileRef::Busy;
+      ref->errorKey  = 0;
+      ref->errorLine = __LINE__;
+      break;
+    }
+
     Ptr<File> file_ptr;
     if (!c_file_hash.find(file_ptr, objId))
     {
@@ -13636,6 +13676,9 @@ Dbdict::execDROP_FILE_REQ(Signal* signal)
     tmp.init<CreateObjRef>(rg, GSN_DROP_OBJ_REF, trans_key);
     sendSignal(rg, GSN_DROP_OBJ_REQ, signal, 
 	       DropObjReq::SignalLength, JBB);
+
+    c_blockState = BS_CREATE_TAB;
+
     return;
   } while(0);
   
@@ -13663,6 +13706,22 @@ Dbdict::execDROP_FILEGROUP_REQ(Signal* signal)
   Uint32 version = req->filegroup_version;
   
   do {
+    if(getOwnNodeId() != c_masterNodeId){
+      jam();
+      ref->errorCode = DropFilegroupRef::NotMaster;
+      ref->errorKey  = 0;
+      ref->errorLine = __LINE__;
+      break;
+    }
+    
+    if (c_blockState != BS_IDLE){
+      jam();
+      ref->errorCode = DropFilegroupRef::Busy;
+      ref->errorKey  = 0;
+      ref->errorLine = __LINE__;
+      break;
+    }
+    
     Ptr<Filegroup> filegroup_ptr;
     if (!c_filegroup_hash.find(filegroup_ptr, objId))
     {
@@ -13718,6 +13777,9 @@ Dbdict::execDROP_FILEGROUP_REQ(Signal* signal)
     tmp.init<CreateObjRef>(rg, GSN_DROP_OBJ_REF, trans_key);
     sendSignal(rg, GSN_DROP_OBJ_REQ, signal, 
 	       DropObjReq::SignalLength, JBB);
+
+    c_blockState = BS_CREATE_TAB;
+
     return;
   } while(0);
   
@@ -13892,6 +13954,7 @@ Dbdict::trans_commit_complete_done(Signal* signal,
     //@todo check api failed
     sendSignal(trans_ptr.p->m_senderRef, GSN_CREATE_FILEGROUP_CONF, signal, 
 	       CreateFilegroupConf::SignalLength, JBB);
+
     break;
   }
   case GSN_CREATE_FILE_REQ:{
@@ -13935,6 +13998,7 @@ Dbdict::trans_commit_complete_done(Signal* signal,
   }
   
   c_Trans.release(trans_ptr);
+  ndbrequire(c_blockState == BS_CREATE_TAB);
   c_blockState = BS_IDLE;
   return;
 }
@@ -14047,6 +14111,7 @@ Dbdict::trans_abort_complete_done(Signal* signal,
   }
   
   c_Trans.release(trans_ptr);
+  ndbrequire(c_blockState == BS_CREATE_TAB);
   c_blockState = BS_IDLE;
   return;
 }
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
index de497ed2c2d..ff917c8482d 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
@@ -1382,8 +1382,9 @@ int Dbtup::handleInsertReq(Signal* signal,
 			regOperPtr.p->userpointer,
 			&regOperPtr.p->m_tuple_location);
     
-    ((Tuple_header*)ptr)->m_operation_ptr_i= regOperPtr.i;
-    ((Tuple_header*)ptr)->m_header_bits= Tuple_header::ALLOC | 
+    base = (Tuple_header*)ptr;
+    base->m_operation_ptr_i= regOperPtr.i;
+    base->m_header_bits= Tuple_header::ALLOC | 
       (varsize ? Tuple_header::CHAINED_ROW : 0);
     regOperPtr.p->m_tuple_location.m_page_no = real_page_id;
   }
@@ -1471,7 +1472,7 @@ int Dbtup::handleInsertReq(Signal* signal,
 size_change_error:
   jam();
   terrorCode = ZMEM_NOMEM_ERROR;
-  goto disk_prealloc_error;
+  goto exit_error;
   
 undo_buffer_error:
   jam();
@@ -1505,9 +1506,13 @@ update_error:
     regOperPtr.p->op_struct.in_active_list = false;
     regOperPtr.p->m_tuple_location.setNull();
   }
-disk_prealloc_error:
+exit_error:
   tupkeyErrorLab(signal);
   return -1;
+
+disk_prealloc_error:
+  base->m_header_bits |= Tuple_header::FREED;
+  goto exit_error;
 }
 
 /* ---------------------------------------------------------------- */
diff --git a/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index c7ef439e193..4808d22b9e0 100644
--- a/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -4410,7 +4410,7 @@ NdbDictInterface::create_file(const NdbFileImpl & file,
   ptr[0].p = (Uint32*)m_buffer.get_data();
   ptr[0].sz = m_buffer.length() / 4;
 
-  int err[] = { CreateFileRef::Busy, 0};
+  int err[] = { CreateFileRef::Busy, CreateFileRef::NotMaster, 0};
   /*
     Send signal without time-out since creating files can take a very long
     time if the file is very big.
@@ -4454,7 +4454,7 @@ NdbDictInterface::drop_file(const NdbFileImpl & file){
   req->file_id = file.m_id;
   req->file_version = file.m_version;
 
-  int err[] = { DropFileRef::Busy, 0};
+  int err[] = { DropFileRef::Busy, DropFileRef::NotMaster, 0};
   DBUG_RETURN(dictSignal(&tSignal, 0, 0,
 	                 0, // master
 		         WAIT_CREATE_INDX_REQ,