Merge orca.ndb.mysql.com:/export/home/space/pekka/ndb/version/my50-bug21017

into  orca.ndb.mysql.com:/export/home/space/pekka/ndb/version/my51-bug21017

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

@@ -192,6 +192,7 @@ public:
   enum ErrorCode {
     NoError = 0,
     Busy = 701,
+    BusyWithNR = 711,
     NotMaster = 702,
     IndexOnDiskAttributeError = 756,
     TriggerNotFound = 4238,

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

@@ -168,6 +168,7 @@ public:
     NoError = 0,
     InvalidIndexVersion = 241,
     Busy = 701,
+    BusyWithNR = 711,
     NotMaster = 702,
     IndexNotFound = 4243,
     BadRequestType = 4247,

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

@@ -7584,9 +7584,18 @@ Dbdict::execCREATE_INDX_REQ(Signal* signal)
     }
     if (signal->getLength() == CreateIndxReq::SignalLength) {
       jam();
+      CreateIndxRef::ErrorCode tmperr = CreateIndxRef::NoError;
       if (getOwnNodeId() != c_masterNodeId) {
         jam();
-	
+        tmperr = CreateIndxRef::NotMaster;
+      } else if (c_blockState == BS_NODE_RESTART) {
+        jam();
+        tmperr = CreateIndxRef::BusyWithNR;
+      } else if (c_blockState != BS_IDLE) {
+        jam();
+        tmperr = CreateIndxRef::Busy;
+      }
+      if (tmperr != CreateIndxRef::NoError) {
 	releaseSections(signal);
 	OpCreateIndex opBusy;
 	opPtr.p = &opBusy;
@@ -7594,13 +7603,12 @@ Dbdict::execCREATE_INDX_REQ(Signal* signal)
 	opPtr.p->m_isMaster = (senderRef == reference());
 	opPtr.p->key = 0;
 	opPtr.p->m_requestType = CreateIndxReq::RT_DICT_PREPARE;
-	opPtr.p->m_errorCode = CreateIndxRef::NotMaster;
+	opPtr.p->m_errorCode = tmperr;
 	opPtr.p->m_errorLine = __LINE__;
 	opPtr.p->m_errorNode = c_masterNodeId;
 	createIndex_sendReply(signal, opPtr, true);
 	return;
       }
-      
       // forward initial request plus operation key to all
       req->setOpKey(++c_opRecordSequence);
       NodeReceiverGroup rg(DBDICT, c_aliveNodes);
@@ -8185,10 +8193,19 @@ Dbdict::execDROP_INDX_REQ(Signal* signal)
     jam();
     if (signal->getLength() == DropIndxReq::SignalLength) {
       jam();
+      DropIndxRef::ErrorCode tmperr = DropIndxRef::NoError;
       if (getOwnNodeId() != c_masterNodeId) {
         jam();
-
-	err = DropIndxRef::NotMaster;
+        tmperr = DropIndxRef::NotMaster;
+      } else if (c_blockState == BS_NODE_RESTART) {
+        jam();
+        tmperr = DropIndxRef::BusyWithNR;
+      } else if (c_blockState != BS_IDLE) {
+        jam();
+        tmperr = DropIndxRef::Busy;
+      }
+      if (tmperr != DropIndxRef::NoError) {
+	err = tmperr;
 	goto error;
       }
       // forward initial request plus operation key to all
@@ -13491,6 +13508,17 @@ Dbdict::execDICT_LOCK_REQ(Signal* signal)
     sendDictLockInfoEvent(lockPtr, "lock request by node");
 }
 
+// only table and index ops are checked
+bool
+Dbdict::hasDictLockSchemaOp()
+{
+  return
+    ! c_opCreateTable.isEmpty() ||
+    ! c_opDropTable.isEmpty() ||
+    ! c_opCreateIndex.isEmpty() ||
+    ! c_opDropIndex.isEmpty();
+}
+
 void
 Dbdict::checkDictLockQueue(Signal* signal, bool poll)
 {
@@ -13511,7 +13539,7 @@ Dbdict::checkDictLockQueue(Signal* signal, bool poll)
       break;
     }
 
-    if (c_opRecordPool.getNoOfFree() != c_opRecordPool.getSize()) {
+    if (hasDictLockSchemaOp()) {
       jam();
       break;
     }
@@ -13544,7 +13572,7 @@ Dbdict::execDICT_UNLOCK_ORD(Signal* signal)
   if (lockPtr.p->locked) {
     jam();
     ndbrequire(c_blockState == lockPtr.p->lt->blockState);
-    ndbrequire(c_opRecordPool.getNoOfFree() == c_opRecordPool.getSize());
+    ndbrequire(! hasDictLockSchemaOp());
     ndbrequire(! c_dictLockQueue.hasPrev(lockPtr));
 
     c_blockState = BS_IDLE;
@@ -13640,7 +13668,7 @@ Dbdict::removeStaleDictLocks(Signal* signal, const Uint32* theFailedNodes)
       if (lockPtr.p->locked) {
         jam();
         ndbrequire(c_blockState == lockPtr.p->lt->blockState);
-        ndbrequire(c_opRecordPool.getNoOfFree() == c_opRecordPool.getSize());
+        ndbrequire(! hasDictLockSchemaOp());
         ndbrequire(! c_dictLockQueue.hasPrev(lockPtr));
 
         c_blockState = BS_IDLE;

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

@@ -2107,6 +2107,9 @@ private:
   void sendDictLockInfoEvent(Uint32 pollCount);
   void sendDictLockInfoEvent(DictLockPtr lockPtr, const char* text);
 
+  // check if any schema op exists (conflicting with dict lock)
+  bool hasDictLockSchemaOp();
+
   void checkDictLockQueue(Signal* signal, bool poll);
   void sendDictLockConf(Signal* signal, DictLockPtr lockPtr);
   void sendDictLockRef(Signal* signal, DictLockReq req, Uint32 errorCode);

storage/ndb/src/kernel/vm/DLHashTable2.hpp

@@ -147,6 +147,8 @@ public:
    */
   bool next(Uint32 bucket, Iterator & iter) const;
 
+  inline bool isEmpty() const { Iterator iter; return ! first(iter); }
+  
 private:
   Uint32 mask;
   Uint32 * hashValues;

storage/ndb/test/include/NDBT_Tables.hpp

@@ -44,6 +44,8 @@ public:
   static const NdbDictionary::Table* getTable(int _num);
   static int getNumTables();
 
+  static const char** getIndexes(const char* table);
+
   static int create_default_tablespace(Ndb* pNdb);
 
 private:

storage/ndb/test/ndbapi/testDict.cpp

@@ -1022,8 +1022,8 @@ int verifyTablesAreEqual(const NdbDictionary::Table* pTab, const NdbDictionary::
   
   if (!pTab->equal(*pTab2)){
     g_err << "equal failed" << endl;
-    g_info << *pTab;
-    g_info << *pTab2;
+    g_info << *(NDBT_Table*)pTab; // gcc-4.1.2
+    g_info << *(NDBT_Table*)pTab2;
     return NDBT_FAILED;
   }
   return NDBT_OK;
@@ -1033,7 +1033,7 @@ int runGetPrimaryKey(NDBT_Context* ctx, NDBT_Step* step){
   Ndb* pNdb = GETNDB(step);
   const NdbDictionary::Table* pTab = ctx->getTab();
   ndbout << "|- " << pTab->getName() << endl;
-  g_info << *pTab;
+  g_info << *(NDBT_Table*)pTab;
   // Try to create table in db
   if (pTab->createTableInDb(pNdb) != 0){
     return NDBT_FAILED;
@@ -1985,7 +1985,53 @@ runDictOps(NDBT_Context* ctx, NDBT_Step* step)
     // replace by the Retrieved table
     pTab = pTab2;
 
-    int records = ctx->getNumRecords();
+    // create indexes
+    const char** indlist = NDBT_Tables::getIndexes(tabName);
+    uint indnum = 0;
+    while (*indlist != 0) {
+      uint count = 0;
+    try_create_index:
+      count++;
+      if (count == 1)
+        g_info << "2: create index " << indnum << " " << *indlist << endl;
+      NdbDictionary::Index ind;
+      char indName[200];
+      sprintf(indName, "%s_X%u", tabName, indnum);
+      ind.setName(indName);
+      ind.setTable(tabName);
+      if (strcmp(*indlist, "UNIQUE") == 0) {
+        ind.setType(NdbDictionary::Index::UniqueHashIndex);
+        ind.setLogging(pTab->getLogging());
+      } else if (strcmp(*indlist, "ORDERED") == 0) {
+        ind.setType(NdbDictionary::Index::OrderedIndex);
+        ind.setLogging(false);
+      } else {
+        assert(false);
+      }
+      const char** indtemp = indlist;
+      while (*++indtemp != 0) {
+        ind.addColumn(*indtemp);
+      }
+      if (pDic->createIndex(ind) != 0) {
+        const NdbError err = pDic->getNdbError();
+        if (count == 1)
+          g_err << "2: " << indName << ": create failed: " << err << endl;
+        if (err.code != 711) {
+          result = NDBT_FAILED;
+          break;
+        }
+        NdbSleep_MilliSleep(myRandom48(maxsleep));
+        goto try_create_index;
+      }
+      indlist = ++indtemp;
+      indnum++;
+    }
+    if (result == NDBT_FAILED)
+      break;
+
+    uint indcount = indnum;
+
+    int records = myRandom48(ctx->getNumRecords());
     g_info << "2: load " << records << " records" << endl;
     HugoTransactions hugoTrans(*pTab);
     if (hugoTrans.loadTable(pNdb, records) != 0) {
@@ -1996,6 +2042,32 @@ runDictOps(NDBT_Context* ctx, NDBT_Step* step)
     }
     NdbSleep_MilliSleep(myRandom48(maxsleep));
 
+    // drop indexes
+    indnum = 0;
+    while (indnum < indcount) {
+      uint count = 0;
+    try_drop_index:
+      count++;
+      if (count == 1)
+        g_info << "2: drop index " << indnum << endl;
+      char indName[200];
+      sprintf(indName, "%s_X%u", tabName, indnum);
+      if (pDic->dropIndex(indName, tabName) != 0) {
+        const NdbError err = pDic->getNdbError();
+        if (count == 1)
+          g_err << "2: " << indName << ": drop failed: " << err << endl;
+        if (err.code != 711) {
+          result = NDBT_FAILED;
+          break;
+        }
+        NdbSleep_MilliSleep(myRandom48(maxsleep));
+        goto try_drop_index;
+      }
+      indnum++;
+    }
+    if (result == NDBT_FAILED)
+      break;
+
     g_info << "2: drop" << endl;
     {
       uint count = 0;

storage/ndb/test/src/NDBT_Tables.cpp

@@ -829,6 +829,17 @@ NDBT_Tables::getNumTables(){
   return numTestTables;
 }
 
+const char**
+NDBT_Tables::getIndexes(const char* table)
+{
+  Uint32 i = 0;
+  for (i = 0; indexes[i].m_table != 0; i++) {
+    if (strcmp(indexes[i].m_table, table) == 0)
+      return indexes[i].m_indexes;
+  }
+  return 0;
+}
+
 int
 NDBT_Tables::createAllTables(Ndb* pNdb, bool _temp, bool existsOk){