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

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

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

@@ -128,7 +128,11 @@ public:
     DihAllAllowNodeStart = 7016,
     DihMinTimeBetweenLCP = 7017,
     DihMaxTimeBetweenLCP = 7018,
+    // 7019
+    // 7020
+    // 7021
     EnableUndoDelayDataWrite = 7080, // DIH+ACC+TUP
+    DihSetTimeBetweenGcp = 7090,
     DihStartLcpImmediately = 7099,
     // 8000 Suma
     // 12000 Tux

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

@@ -64,5 +64,11 @@ private:
   
   Uint32 startingNodeId;
   Uint32 errorCode;  
+
+  enum ErrorCode
+  {
+    ZNODE_ALREADY_STARTING_ERROR = 305,
+    InitialStartRequired = 320
+  };
 };
 #endif

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

@@ -44,7 +44,8 @@ private:
     CHECK_WAIT_DROP_TAB_FAILED_LQH         = 16,
     TRIGGER_PENDING                        = 17,
     
-    DelayTCKEYCONF = 18
+    DelayTCKEYCONF = 18,
+    ZNF_CHECK_TRANSACTIONS = 19
   };
 };
 

storage/ndb/include/mgmapi/ndbd_exit_codes.h

@@ -74,9 +74,7 @@ typedef ndbd_exit_classification_enum ndbd_exit_classification;
 #define NDBD_EXIT_SR_OTHERNODEFAILED          2308
 #define NDBD_EXIT_NODE_NOT_DEAD               2309
 #define NDBD_EXIT_SR_REDOLOG                  2310
-/*
 #define NDBD_EXIT_SR_RESTARTCONFLICT          2311
-*/
 #define NDBD_EXIT_NO_MORE_UNDOLOG             2312 
 #define NDBD_EXIT_SR_UNDOLOG                  2313 
 #define NDBD_EXIT_MEMALLOC                    2327

storage/ndb/src/kernel/blocks/ERROR_codes.txt

@@ -228,6 +228,8 @@ Delay execution of COMPLETECONF signal 2 seconds to generate time-out.
 8045: (ABORTCONF only as part of take-over)
 Delay execution of ABORTCONF signal 2 seconds to generate time-out.
 
+8050: Send ZABORT_TIMEOUT_BREAK delayed
+
 ERROR CODES FOR TESTING TIME-OUT HANDLING IN DBTC
 -------------------------------------------------
 
@@ -305,6 +307,8 @@ Test Crashes in handling node restarts
 7131: Crash when receiving START_COPYREQ in master node
 7132: Crash when receiving START_COPYCONF in starting node
 
+7170: Crash when receiving START_PERMREF (InitialStartRequired)
+
 DICT:
 6000  Crash during NR when receiving DICTSTARTREQ
 6001  Crash during NR when receiving SCHEMA_INFO
@@ -314,7 +318,7 @@ LQH:
 5026  Crash when receiving COPY_ACTIVEREQ
 5027  Crash when receiving STAT_RECREQ
 
-5042  Crash starting node, when scan is finished on primary replica
+5043  Crash starting node, when scan is finished on primary replica
 
 Test Crashes in handling take over
 ----------------------------------

storage/ndb/src/kernel/blocks/dbdih/Dbdih.hpp

@@ -75,7 +75,6 @@
 #define ZWRONG_FAILURE_NUMBER_ERROR 302
 #define ZWRONG_START_NODE_ERROR 303
 #define ZNO_REPLICA_FOUND_ERROR 304
-#define ZNODE_ALREADY_STARTING_ERROR 305
 #define ZNODE_START_DISALLOWED_ERROR 309
 
 // --------------------------------------
@@ -1040,7 +1039,8 @@ private:
   void prepareReplicas(FragmentstorePtr regFragptr);
   void removeNodeFromStored(Uint32 nodeId,
                             FragmentstorePtr regFragptr,
-                            ReplicaRecordPtr replicaPtr);
+                            ReplicaRecordPtr replicaPtr,
+			    bool temporary);
   void removeOldStoredReplica(FragmentstorePtr regFragptr,
                               ReplicaRecordPtr replicaPtr);
   void removeStoredReplica(FragmentstorePtr regFragptr,

storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp

@@ -9960,7 +9960,7 @@ void Dblqh::nextScanConfCopyLab(Signal* signal)
 /*---------------------------------------------------------------------------*/
     scanptr.p->scanCompletedStatus = ZTRUE;
     scanptr.p->scanState = ScanRecord::WAIT_LQHKEY_COPY;
-    if (ERROR_INSERTED(5042))
+    if (ERROR_INSERTED(5043))
     {
       CLEAR_ERROR_INSERT_VALUE;
       tcConnectptr.p->copyCountWords = ~0;
@@ -18445,6 +18445,172 @@ Dblqh::execDUMP_STATE_ORD(Signal* signal)
     c_error_insert_table_id = dumpState->args[1];
     SET_ERROR_INSERT_VALUE(5042);
   }
+
+  TcConnectionrec *regTcConnectionrec = tcConnectionrec;
+  Uint32 ttcConnectrecFileSize = ctcConnectrecFileSize;
+  if(arg == 2306)
+  {
+    for(Uint32 i = 0; i<1024; i++)
+    {
+      TcConnectionrecPtr tcRec;
+      tcRec.i = ctransidHash[i];
+      while(tcRec.i != RNIL)
+      {
+	ptrCheckGuard(tcRec, ttcConnectrecFileSize, regTcConnectionrec);
+	ndbout << "TcConnectionrec " << tcRec.i;
+	signal->theData[0] = 2307;
+	signal->theData[1] = tcRec.i;
+	execDUMP_STATE_ORD(signal);
+	tcRec.i = tcRec.p->nextHashRec;
+      }
+    }
+  }
+
+  if(arg == 2307 || arg == 2308)
+  {
+    TcConnectionrecPtr tcRec;
+    tcRec.i = signal->theData[1];
+    ptrCheckGuard(tcRec, ttcConnectrecFileSize, regTcConnectionrec);
+    
+    ndbout << " transactionState = " << tcRec.p->transactionState<<endl;
+    ndbout << " operation = " << tcRec.p->operation<<endl;
+    ndbout << " tcNodeFailrec = " << tcRec.p->tcNodeFailrec
+	   << " seqNoReplica = " << tcRec.p->seqNoReplica
+	   << " simpleRead = " << tcRec.p->simpleRead
+	   << endl;
+    ndbout << " replicaType = " << tcRec.p->replicaType
+	   << " reclenAiLqhkey = " << tcRec.p->reclenAiLqhkey
+	   << " opExec = " << tcRec.p->opExec
+	   << endl;
+    ndbout << " opSimple = " << tcRec.p->opSimple
+	   << " nextSeqNoReplica = " << tcRec.p->nextSeqNoReplica
+	   << " lockType = " << tcRec.p->lockType
+	   << endl;
+    ndbout << " lastReplicaNo = " << tcRec.p->lastReplicaNo
+	   << " indTakeOver = " << tcRec.p->indTakeOver
+	   << " dirtyOp = " << tcRec.p->dirtyOp
+	   << endl;
+    ndbout << " activeCreat = " << tcRec.p->activeCreat
+	   << " tcBlockref = " << hex << tcRec.p->tcBlockref
+	   << " reqBlockref = " << hex << tcRec.p->reqBlockref
+	   << " primKeyLen = " << tcRec.p->primKeyLen
+	   << endl;
+    ndbout << " nextReplica = " << tcRec.p->nextReplica
+	   << " tcBlockref = " << hex << tcRec.p->tcBlockref
+	   << " reqBlockref = " << hex << tcRec.p->reqBlockref
+	   << " primKeyLen = " << tcRec.p->primKeyLen
+	   << endl;
+    ndbout << " logStopPageNo = " << tcRec.p->logStopPageNo
+	   << " logStartPageNo = " << tcRec.p->logStartPageNo
+	   << " logStartPageIndex = " << tcRec.p->logStartPageIndex
+	   << endl;
+    ndbout << " errorCode = " << tcRec.p->errorCode
+	   << " clientBlockref = " << hex << tcRec.p->clientBlockref
+	   << " applRef = " << hex << tcRec.p->applRef
+	   << " totSendlenAi = " << tcRec.p->totSendlenAi
+	   << endl;
+    ndbout << " totReclenAi = " << tcRec.p->totReclenAi
+	   << " tcScanRec = " << tcRec.p->tcScanRec
+	   << " tcScanInfo = " << tcRec.p->tcScanInfo
+	   << " tcOprec = " << hex << tcRec.p->tcOprec
+	   << endl;
+    ndbout << " tableref = " << tcRec.p->tableref
+	   << " simpleTcConnect = " << tcRec.p->simpleTcConnect
+	   << " storedProcId = " << tcRec.p->storedProcId
+	   << " schemaVersion = " << tcRec.p->schemaVersion
+	   << endl;
+    ndbout << " reqinfo = " << tcRec.p->reqinfo
+	   << " reqRef = " << tcRec.p->reqRef
+	   << " readlenAi = " << tcRec.p->readlenAi
+	   << " prevTc = " << tcRec.p->prevTc
+	   << endl;
+    ndbout << " prevLogTcrec = " << tcRec.p->prevLogTcrec
+	   << " prevHashRec = " << tcRec.p->prevHashRec
+	   << " nodeAfterNext0 = " << tcRec.p->nodeAfterNext[0]
+	   << " nodeAfterNext1 = " << tcRec.p->nodeAfterNext[1]
+	   << endl;
+    ndbout << " nextTcConnectrec = " << tcRec.p->nextTcConnectrec
+	   << " nextTc = " << tcRec.p->nextTc
+	   << " nextTcLogQueue = " << tcRec.p->nextTcLogQueue
+	   << " nextLogTcrec = " << tcRec.p->nextLogTcrec
+	   << endl;
+    ndbout << " nextHashRec = " << tcRec.p->nextHashRec
+	   << " logWriteState = " << tcRec.p->logWriteState
+	   << " logStartFileNo = " << tcRec.p->logStartFileNo
+	   << " listState = " << tcRec.p->listState
+	   << endl;
+    ndbout << " lastAttrinbuf = " << tcRec.p->lastAttrinbuf
+	   << " lastTupkeybuf = " << tcRec.p->lastTupkeybuf
+	   << " hashValue = " << tcRec.p->hashValue
+	   << endl;
+    ndbout << " gci = " << tcRec.p->gci
+	   << " fragmentptr = " << tcRec.p->fragmentptr
+	   << " fragmentid = " << tcRec.p->fragmentid
+	   << " firstTupkeybuf = " << tcRec.p->firstTupkeybuf
+	   << endl;
+    ndbout << " firstAttrinbuf = " << tcRec.p->firstAttrinbuf
+	   << " currTupAiLen = " << tcRec.p->currTupAiLen
+	   << " currReclenAi = " << tcRec.p->currReclenAi
+	   << endl;
+    ndbout << " tcTimer = " << tcRec.p->tcTimer
+	   << " clientConnectrec = " << tcRec.p->clientConnectrec
+	   << " applOprec = " << hex << tcRec.p->applOprec
+	   << " abortState = " << tcRec.p->abortState
+	   << endl;
+    ndbout << " transid0 = " << hex << tcRec.p->transid[0]
+	   << " transid1 = " << hex << tcRec.p->transid[1]
+	   << " tupkeyData0 = " << tcRec.p->tupkeyData[0]
+	   << " tupkeyData1 = " << tcRec.p->tupkeyData[1]
+	   << endl;
+    ndbout << " tupkeyData2 = " << tcRec.p->tupkeyData[2]
+	   << " tupkeyData3 = " << tcRec.p->tupkeyData[3]
+	   << endl;
+    switch (tcRec.p->transactionState) {
+	
+    case TcConnectionrec::SCAN_STATE_USED:
+      if (tcRec.p->tcScanRec < cscanrecFileSize){
+	ScanRecordPtr TscanPtr;
+	c_scanRecordPool.getPtr(TscanPtr, tcRec.p->tcScanRec);
+	ndbout << " scanState = " << TscanPtr.p->scanState << endl;
+	//TscanPtr.p->scanLocalref[2];
+	ndbout << " copyPtr="<<TscanPtr.p->copyPtr
+	       << " scanAccPtr="<<TscanPtr.p->scanAccPtr
+	       << " scanAiLength="<<TscanPtr.p->scanAiLength
+	       << endl;
+	ndbout << " m_curr_batch_size_rows="<<
+	  TscanPtr.p->m_curr_batch_size_rows
+	       << " m_max_batch_size_rows="<<
+	  TscanPtr.p->m_max_batch_size_rows
+	       << " scanErrorCounter="<<TscanPtr.p->scanErrorCounter
+	       << endl;
+	ndbout << " scanSchemaVersion="<<TscanPtr.p->scanSchemaVersion
+	       << "  scanStoredProcId="<<TscanPtr.p->scanStoredProcId
+	       << "  scanTcrec="<<TscanPtr.p->scanTcrec
+	       << endl;
+	ndbout << "  scanType="<<TscanPtr.p->scanType
+	       << "  scanApiBlockref="<<TscanPtr.p->scanApiBlockref
+	       << "  scanNodeId="<<TscanPtr.p->scanNodeId
+	       << "  scanCompletedStatus="<<TscanPtr.p->scanCompletedStatus
+	       << endl;
+	ndbout << "  scanFlag="<<TscanPtr.p->scanFlag
+	       << "  scanLockHold="<<TscanPtr.p->scanLockHold
+	       << "  scanLockMode="<<TscanPtr.p->scanLockMode
+	       << "  scanNumber="<<TscanPtr.p->scanNumber
+	       << endl;
+	ndbout << "  scanReleaseCounter="<<TscanPtr.p->scanReleaseCounter
+	       << "  scanTcWaiting="<<TscanPtr.p->scanTcWaiting
+	       << "  scanKeyinfoFlag="<<TscanPtr.p->scanKeyinfoFlag
+	       << endl;
+      } else{
+	ndbout << "No connected scan record found" << endl;
+      }
+      break;
+    default:
+      break;
+    }
+    ndbrequire(arg != 2308);
+  }
+  
 }//Dblqh::execDUMP_STATE_ORD()
 
 void Dblqh::execSET_VAR_REQ(Signal* signal) 

storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp

@@ -213,14 +213,6 @@ public:
     LTS_ACTIVE = 1
   };
 
-  enum TakeOverState {
-    TOS_NOT_DEFINED = 0,
-    TOS_IDLE = 1,
-    TOS_ACTIVE = 2,
-    TOS_COMPLETED = 3,
-    TOS_NODE_FAILED = 4
-  };
-
   enum FailState {
     FS_IDLE = 0,
     FS_LISTENING = 1,
@@ -645,6 +637,7 @@ public:
     ConnectionState apiConnectstate;
     UintR transid[2];
     UintR firstTcConnect;
+    NdbNodeBitmask m_transaction_nodes; 
     
     //---------------------------------------------------
     // Second 16 byte cache line. Hot variables.
@@ -937,7 +930,6 @@ public:
   struct HostRecord {
     HostState hostStatus;
     LqhTransState lqhTransStatus;
-    TakeOverState takeOverStatus;
     bool  inPackedList;
     UintR noOfPackedWordsLqh;
     UintR packedWordsLqh[26];
@@ -946,6 +938,17 @@ public:
     UintR noOfWordsTCINDXCONF;
     UintR packedWordsTCINDXCONF[30];
     BlockReference hostLqhBlockRef;
+
+    enum NodeFailBits
+    {
+      NF_TAKEOVER          = 0x1,
+      NF_CHECK_SCAN        = 0x2,
+      NF_CHECK_TRANSACTION = 0x4,
+      NF_CHECK_DROP_TAB    = 0x8,
+      NF_NODE_FAIL_BITS    = 0xF // All bits...
+    };
+    Uint32 m_nf_bits;
+    NdbNodeBitmask m_lqh_trans_conf;
   }; /* p2c: size = 128 bytes */
   
   typedef Ptr<HostRecord> HostRecordPtr;
@@ -1593,7 +1596,7 @@ private:
   void wrongSchemaVersionErrorLab(Signal* signal);
   void noFreeConnectionErrorLab(Signal* signal);
   void tckeyreq050Lab(Signal* signal);
-  void timeOutFoundLab(Signal* signal, UintR anAdd);
+  void timeOutFoundLab(Signal* signal, UintR anAdd, Uint32 errCode);
   void completeTransAtTakeOverLab(Signal* signal, UintR TtakeOverInd);
   void completeTransAtTakeOverDoLast(Signal* signal, UintR TtakeOverInd);
   void completeTransAtTakeOverDoOne(Signal* signal, UintR TtakeOverInd);
@@ -1615,6 +1618,9 @@ private:
   void checkScanFragList(Signal*, Uint32 failedNodeId, ScanRecord * scanP, 
 			 LocalDLList<ScanFragRec>::Head&);
 
+  void nodeFailCheckTransactions(Signal*,Uint32 transPtrI,Uint32 failedNodeId);
+  void checkNodeFailComplete(Signal* signal, Uint32 failedNodeId, Uint32 bit);
+  
   // Initialisation
   void initData();
   void initRecords();
@@ -1641,6 +1647,7 @@ private:
   HostRecord *hostRecord;
   HostRecordPtr hostptr;
   UintR chostFilesize;
+  NdbNodeBitmask c_alive_nodes;
 
   GcpRecord *gcpRecord;
   GcpRecordPtr gcpPtr;

storage/ndb/src/kernel/blocks/qmgr/QmgrMain.cpp

@@ -278,6 +278,7 @@ void Qmgr::setArbitTimeout(UintR aArbitTimeout)
 
 void Qmgr::execCONNECT_REP(Signal* signal)
 {
+  jamEntry();
   const Uint32 nodeId = signal->theData[0];
   c_connectedNodes.set(nodeId);
   NodeRecPtr nodePtr;
@@ -285,9 +286,13 @@ void Qmgr::execCONNECT_REP(Signal* signal)
   ptrCheckGuard(nodePtr, MAX_NODES, nodeRec);
   switch(nodePtr.p->phase){
   case ZSTARTING:
+  case ZRUNNING:
+    jam();
+    if(!c_start.m_nodes.isWaitingFor(nodeId)){
       jam();
+      return;
+    }
     break;
-  case ZRUNNING:
   case ZPREPARE_FAIL:
   case ZFAIL_CLOSING:
     jam();
@@ -299,20 +304,27 @@ void Qmgr::execCONNECT_REP(Signal* signal)
     return;
   }
   
-  if(!c_start.m_nodes.isWaitingFor(nodeId)){
-    jam();
-    return;
-  }
-
   switch(c_start.m_gsn){
   case GSN_CM_REGREQ:
     jam();
     sendCmRegReq(signal, nodeId);
     return;
-  case GSN_CM_NODEINFOREQ:{
+  case GSN_CM_NODEINFOREQ:
     jam();
     sendCmNodeInfoReq(signal, nodeId, nodePtr.p);
     return;
+  case GSN_CM_ADD:{
+    jam();
+
+    ndbrequire(getOwnNodeId() != cpresident);
+    c_start.m_nodes.clearWaitingFor(nodeId);
+    c_start.m_gsn = RNIL;
+    
+    NodeRecPtr addNodePtr;
+    addNodePtr.i = nodeId;
+    ptrCheckGuard(addNodePtr, MAX_NDB_NODES, nodeRec);
+    cmAddPrepare(signal, addNodePtr, nodePtr.p);
+    return;
   }
   default:
     return;
@@ -945,15 +957,27 @@ Qmgr::cmAddPrepare(Signal* signal, NodeRecPtr nodePtr, const NodeRec * self){
     return;
   case ZFAIL_CLOSING:
     jam();
-#ifdef VM_TRACE
-    ndbout_c("Enabling communication to CM_ADD node state=%d", 
+    
+#if 1
+    warningEvent("Recieved request to incorperate node %u, "
+		 "while error handling has not yet completed",
+		 nodePtr.i);
+    
+    ndbrequire(getOwnNodeId() != cpresident);
+    ndbrequire(signal->header.theVerId_signalNumber == GSN_CM_ADD);
+    c_start.m_nodes.clearWaitingFor();
+    c_start.m_nodes.setWaitingFor(nodePtr.i);
+    c_start.m_gsn = GSN_CM_ADD;
+#else
+    warningEvent("Enabling communication to CM_ADD node %u state=%d", 
+		 nodePtr.i,
 		 nodePtr.p->phase);
-#endif
     nodePtr.p->phase = ZSTARTING;
     nodePtr.p->failState = NORMAL;
     signal->theData[0] = 0;
     signal->theData[1] = nodePtr.i;
     sendSignal(CMVMI_REF, GSN_OPEN_COMREQ, signal, 2, JBA);
+#endif
     return;
   case ZSTARTING:
     break;
@@ -1788,11 +1812,27 @@ void Qmgr::execNDB_FAILCONF(Signal* signal)
 
   jamEntry();
   failedNodePtr.i = signal->theData[0];  
+
+  if (ERROR_INSERTED(930))
+  {
+    CLEAR_ERROR_INSERT_VALUE;
+    infoEvent("Discarding NDB_FAILCONF for %u", failedNodePtr.i);
+    return;
+  }
+  
   ptrCheckGuard(failedNodePtr, MAX_NODES, nodeRec);
   if (failedNodePtr.p->failState == WAITING_FOR_NDB_FAILCONF){
     failedNodePtr.p->failState = NORMAL;
   } else {
     jam();
+
+    char buf[100];
+    BaseString::snprintf(buf, 100, 
+			 "Received NDB_FAILCONF for node %u with state: %d %d",
+			 failedNodePtr.i,
+			 failedNodePtr.p->phase,
+			 failedNodePtr.p->failState);
+    progError(__LINE__, 0, buf);
     systemErrorLab(signal, __LINE__);
   }//if
   if (cpresident == getOwnNodeId()) {
@@ -2115,7 +2155,39 @@ void Qmgr::failReportLab(Signal* signal, Uint16 aFailedNode,
   ptrCheckGuard(failedNodePtr, MAX_NODES, nodeRec);
   if (failedNodePtr.i == getOwnNodeId()) {
     jam();
-    systemErrorLab(signal, __LINE__);
+
+    const char * msg = 0;
+    switch(aFailCause){
+    case FailRep::ZOWN_FAILURE: 
+      msg = "Own failure"; 
+      break;
+    case FailRep::ZOTHER_NODE_WHEN_WE_START: 
+    case FailRep::ZOTHERNODE_FAILED_DURING_START:
+      msg = "Other node died during start"; 
+      break;
+    case FailRep::ZIN_PREP_FAIL_REQ:
+      msg = "Prep fail";
+      break;
+    case FailRep::ZSTART_IN_REGREQ:
+      msg = "Start timeout";
+      break;
+    case FailRep::ZHEARTBEAT_FAILURE:
+      msg = "Hearbeat failure";
+      break;
+    case FailRep::ZLINK_FAILURE:
+      msg = "Connection failure";
+      break;
+    }
+    
+    char buf[100];
+    BaseString::snprintf(buf, 100, 
+			 "We(%u) have been declared dead by %u reason: %s(%u)",
+			 getOwnNodeId(),
+			 refToNode(signal->getSendersBlockRef()),
+			 aFailCause,
+			 msg ? msg : "<Unknown>");
+
+    progError(__LINE__, 0, buf);
     return;
   }//if
   
@@ -2829,6 +2901,7 @@ void Qmgr::failReport(Signal* signal,
         cfailureNr = cprepareFailureNr;
         ctoFailureNr = 0;
         ctoStatus = Q_ACTIVE;
+	c_start.reset(); // Don't take over nodes being started
         if (cnoCommitFailedNodes > 0) {
           jam();
 	  /**-----------------------------------------------------------------

storage/ndb/src/kernel/error/ndbd_exit_codes.c

@@ -85,6 +85,9 @@ static const ErrStruct errArray[] =
    */
    {NDBD_EXIT_OS_SIGNAL_RECEIVED, XIE, "Error OS signal received"},
 
+   {NDBD_EXIT_SR_RESTARTCONFLICT, XRE,
+    "Partial system restart causing conflicting file systems"},
+   
    /* VM */
    {NDBD_EXIT_OUT_OF_LONG_SIGNAL_MEMORY,    XCR,
     "Signal lost, out of long signal memory, please increase LongMessageBuffer"},

storage/ndb/src/kernel/vm/TransporterCallback.cpp

@@ -55,7 +55,9 @@ static const ConnectionError connectionError[] =
 const char *lookupConnectionError(Uint32 err)
 {
   int i= 0;
-  while ((Uint32)connectionError[i].err != err && (Uint32)connectionError[i].err != -1);
+  while ((Uint32)connectionError[i].err != err && 
+	 (Uint32)connectionError[i].err != -1)
+    i++;
   return connectionError[i].text;
 }
 

storage/ndb/src/ndbapi/NdbTransaction.cpp

@@ -434,12 +434,12 @@ NdbTransaction::executeNoBlobs(ExecType aTypeOfExec,
 //------------------------------------------------------------------------
   Ndb* tNdb = theNdb;
 
+  Uint32 timeout = TransporterFacade::instance()->m_waitfor_timeout;
   m_waitForReply = false;
   executeAsynchPrepare(aTypeOfExec, NULL, NULL, abortOption);
   if (m_waitForReply){
     while (1) {
-      int noOfComp = tNdb->sendPollNdb((3 * WAITFOR_RESPONSE_TIMEOUT),
-                                       1, forceSend);
+      int noOfComp = tNdb->sendPollNdb(3 * timeout, 1, forceSend);
       if (noOfComp == 0) {
         /** 
          * This timeout situation can occur if NDB crashes.

storage/ndb/src/ndbapi/Ndbif.cpp

@@ -1055,6 +1055,7 @@ Ndb::pollCompleted(NdbTransaction** aCopyArray)
 void
 Ndb::check_send_timeout()
 {
+  Uint32 timeout = TransporterFacade::instance()->m_waitfor_timeout;
   NDB_TICKS current_time = NdbTick_CurrentMillisecond();
   assert(current_time >= the_last_check_time);
   if (current_time - the_last_check_time > 1000) {
@@ -1062,17 +1063,18 @@ Ndb::check_send_timeout()
     Uint32 no_of_sent = theNoOfSentTransactions;
     for (Uint32 i = 0; i < no_of_sent; i++) {
       NdbTransaction* a_con = theSentTransactionsArray[i];
-      if ((current_time - a_con->theStartTransTime) >
-          WAITFOR_RESPONSE_TIMEOUT) {
+      if ((current_time - a_con->theStartTransTime) > timeout)
+      {
 #ifdef VM_TRACE
         a_con->printState();
 	Uint32 t1 = a_con->theTransactionId;
 	Uint32 t2 = a_con->theTransactionId >> 32;
-	ndbout_c("[%.8x %.8x]", t1, t2);
-	abort();
+	ndbout_c("4012 [%.8x %.8x]", t1, t2);
+	//abort();
 #endif
+        a_con->theReleaseOnClose = true;
         a_con->setOperationErrorCodeAbort(4012);
-        a_con->theCommitStatus = NdbTransaction::Aborted;
+	a_con->theCommitStatus = NdbTransaction::NeedAbort;
         a_con->theCompletionStatus = NdbTransaction::CompletedFailure;
         a_con->handleExecuteCompletion();
         remove_sent_list(i);

storage/ndb/src/ndbapi/TransporterFacade.cpp

@@ -721,6 +721,19 @@ TransporterFacade::init(Uint32 nodeId, const ndb_mgm_configuration* props)
     m_batch_size= batch_size;
   }
   
+  Uint32 timeout = 120000;
+  iter.first();
+  for (iter.first(); iter.valid(); iter.next())
+  {
+    Uint32 tmp1 = 0, tmp2 = 0;
+    iter.get(CFG_DB_TRANSACTION_CHECK_INTERVAL, &tmp1);
+    iter.get(CFG_DB_TRANSACTION_DEADLOCK_TIMEOUT, &tmp2);
+    tmp1 += tmp2;
+    if (tmp1 > timeout)
+      timeout = tmp1;
+  }
+  m_waitfor_timeout = timeout;
+  
   if (!theTransporterRegistry->start_service(m_socket_server)){
     ndbout_c("Unable to start theTransporterRegistry->start_service");
     DBUG_RETURN(false);

storage/ndb/src/ndbapi/TransporterFacade.hpp

@@ -121,6 +121,7 @@ public:
   Uint32 get_scan_batch_size();
   Uint32 get_batch_byte_size();
   Uint32 get_batch_size();
+  Uint32 m_waitfor_timeout; // in milli seconds...
 
   TransporterRegistry* get_registry() { return theTransporterRegistry;};
 

storage/ndb/test/include/NdbRestarter.hpp

@@ -62,6 +62,7 @@ public:
   int dumpStateAllNodes(int * _args, int _num_args);
 
   int getMasterNodeId();
+  int getRandomNodeSameNodeGroup(int nodeId, int randomNumber);
   int getRandomNodeOtherNodeGroup(int nodeId, int randomNumber);
   int getRandomNotMasterNodeId(int randomNumber);
   

storage/ndb/test/ndbapi/testNodeRestart.cpp

@@ -439,6 +439,14 @@ int runBug15587(NDBT_Context* ctx, NDBT_Step* step){
   if (restarter.startNodes(&nodeId, 1))
     return NDBT_FAILED;
 
+  restarter.waitNodesStartPhase(&nodeId, 1, 3);
+  
+  if (restarter.waitNodesNoStart(&nodeId, 1))
+    return NDBT_FAILED; 
+   
+  if (restarter.startNodes(&nodeId, 1))
+    return NDBT_FAILED;
+  
   if (restarter.waitNodesStarted(&nodeId, 1))
     return NDBT_FAILED;
   
@@ -535,6 +543,119 @@ int runBug15685(NDBT_Context* ctx, NDBT_Step* step){
   return NDBT_FAILED;
 }
 
+int 
+runBug16772(NDBT_Context* ctx, NDBT_Step* step){
+
+  NdbRestarter restarter;
+  if (restarter.getNumDbNodes() < 2)
+  {
+    ctx->stopTest();
+    return NDBT_OK;
+  }
+
+  int aliveNodeId = restarter.getRandomNotMasterNodeId(rand());
+  int deadNodeId = aliveNodeId;
+  while (deadNodeId == aliveNodeId)
+    deadNodeId = restarter.getDbNodeId(rand() % restarter.getNumDbNodes());
+  
+  if (restarter.insertErrorInNode(aliveNodeId, 930))
+    return NDBT_FAILED;
+
+  if (restarter.restartOneDbNode(deadNodeId,
+				 /** initial */ false, 
+				 /** nostart */ true,
+				 /** abort   */ true))
+    return NDBT_FAILED;
+  
+  if (restarter.waitNodesNoStart(&deadNodeId, 1))
+    return NDBT_FAILED;
+
+  if (restarter.startNodes(&deadNodeId, 1))
+    return NDBT_FAILED;
+
+  // It should now be hanging since we throw away NDB_FAILCONF
+  int ret = restarter.waitNodesStartPhase(&deadNodeId, 1, 3, 10);
+  // So this should fail...i.e it should not reach startphase 3
+
+  // Now send a NDB_FAILCONF for deadNo
+  int dump[] = { 7020, 323, 252, 0 };
+  dump[3] = deadNodeId;
+  if (restarter.dumpStateOneNode(aliveNodeId, dump, 4))
+    return NDBT_FAILED;
+  
+  if (restarter.waitNodesStarted(&deadNodeId, 1))
+    return NDBT_FAILED;
+
+  return ret ? NDBT_OK : NDBT_FAILED;
+}
+
+int 
+runBug18414(NDBT_Context* ctx, NDBT_Step* step){
+
+  NdbRestarter restarter;
+  if (restarter.getNumDbNodes() < 2)
+  {
+    ctx->stopTest();
+    return NDBT_OK;
+  }
+
+  Ndb* pNdb = GETNDB(step);
+  HugoOperations hugoOps(*ctx->getTab());
+  HugoTransactions hugoTrans(*ctx->getTab());
+  int loop = 0;
+  do 
+  {
+    if(hugoOps.startTransaction(pNdb) != 0)
+      goto err;
+    
+    if(hugoOps.pkUpdateRecord(pNdb, 0, 128, rand()) != 0)
+      goto err;
+    
+    if(hugoOps.execute_NoCommit(pNdb) != 0)
+      goto err;
+
+    int node1 = hugoOps.getTransaction()->getConnectedNodeId();
+    int node2 = restarter.getRandomNodeSameNodeGroup(node1, rand());
+    
+    if (node1 == -1 || node2 == -1)
+      break;
+    
+    if (loop & 1)
+    {
+      if (restarter.insertErrorInNode(node1, 8050))
+	goto err;
+    }
+    
+    if (restarter.insertErrorInNode(node2, 5003))
+      goto err;
+    
+    int res= hugoOps.execute_Rollback(pNdb);
+  
+    if (restarter.waitNodesNoStart(&node2, 1) != 0)
+      goto err;
+    
+    if (restarter.insertErrorInAllNodes(0))
+      goto err;
+    
+    if (restarter.startNodes(&node2, 1) != 0)
+      goto err;
+    
+    if (restarter.waitClusterStarted() != 0)
+      goto err;
+    
+    if (hugoTrans.scanUpdateRecords(pNdb, 128) != 0)
+      goto err;
+
+    hugoOps.closeTransaction(pNdb);
+    
+  } while(++loop < 5);
+  
+  return NDBT_OK;
+  
+err:
+  hugoOps.closeTransaction(pNdb);
+  return NDBT_FAILED;    
+}
 
 NDBT_TESTSUITE(testNodeRestart);
 TESTCASE("NoLoad", 
@@ -820,6 +941,16 @@ TESTCASE("Bug15685",
   STEP(runBug15685);
   FINALIZER(runClearTable);
 }
+TESTCASE("Bug16772",
+	 "Test bug with restarting before NF handling is complete"){
+  STEP(runBug16772);
+}
+TESTCASE("Bug18414",
+	 "Test bug with NF during NR"){
+  INITIALIZER(runLoadTable);
+  STEP(runBug18414);
+  FINALIZER(runClearTable);
+}
 NDBT_TESTSUITE_END(testNodeRestart);
 
 int main(int argc, const char** argv){

storage/ndb/test/ndbapi/testSystemRestart.cpp

@@ -1051,6 +1051,52 @@ int runSystemRestart9(NDBT_Context* ctx, NDBT_Step* step){
   return result;
 }
 
+int runBug18385(NDBT_Context* ctx, NDBT_Step* step){
+  NdbRestarter restarter;
+  const Uint32 nodeCount = restarter.getNumDbNodes();
+  if(nodeCount < 2){
+    g_info << "Bug18385 - Needs atleast 2 nodes to test" << endl;
+    return NDBT_OK;
+  }
+
+  int node1 = restarter.getDbNodeId(rand() % nodeCount);
+  int node2 = restarter.getRandomNodeSameNodeGroup(node1, rand());
+
+  if (node1 == -1 || node2 == -1)
+    return NDBT_OK;
+  
+  int dump[] = { DumpStateOrd::DihSetTimeBetweenGcp, 300 };
+  
+  int result = NDBT_OK;
+  do {
+    CHECK(restarter.dumpStateAllNodes(dump, 2) == 0);
+    CHECK(restarter.restartOneDbNode(node1, false, true, false) == 0);
+    NdbSleep_SecSleep(3);
+    CHECK(restarter.restartAll(false, true, false) == 0);
+    
+    Uint32 cnt = 0;
+    int nodes[128];
+    for(Uint32 i = 0; i<nodeCount; i++)
+      if ((nodes[cnt] = restarter.getDbNodeId(i)) != node2)
+	cnt++;
+    
+    assert(cnt == nodeCount - 1);
+    
+    CHECK(restarter.startNodes(nodes, cnt) == 0);
+    CHECK(restarter.waitNodesStarted(nodes, cnt, 300) == 0);
+    
+    CHECK(restarter.insertErrorInNode(node2, 7170) == 0);
+    CHECK(restarter.waitNodesNoStart(&node2, 1) == 0);
+    CHECK(restarter.restartOneDbNode(node2, true, false, true) == 0);
+    CHECK(restarter.waitNodesStarted(&node2, 1) == 0);
+
+  } while(0);
+  
+  g_info << "Bug18385 finished" << endl;  
+  
+  return result;
+}
+
 int runWaitStarted(NDBT_Context* ctx, NDBT_Step* step){
 
   NdbRestarter restarter;
@@ -1234,6 +1280,13 @@ TESTCASE("SR9",
   STEP(runSystemRestart9);
   FINALIZER(runClearTable);
 }
+TESTCASE("Bug18385", 
+	 "Perform partition system restart with other nodes with higher GCI"){
+  INITIALIZER(runWaitStarted);
+  INITIALIZER(runClearTable);
+  STEP(runBug18385);
+  FINALIZER(runClearTable);
+}
 NDBT_TESTSUITE_END(testSystemRestart);
 
 int main(int argc, const char** argv){

storage/ndb/test/ndbapi/testTimeout.cpp

@@ -24,6 +24,7 @@
 
 #define TIMEOUT (Uint32)3000
 Uint32 g_org_timeout = 3000;
+Uint32 g_org_deadlock = 3000;
 
 int
 setTransactionTimeout(NDBT_Context* ctx, NDBT_Step* step){
@@ -59,6 +60,60 @@ resetTransactionTimeout(NDBT_Context* ctx, NDBT_Step* step){
   return NDBT_OK;
 }
 
+int
+setDeadlockTimeout(NDBT_Context* ctx, NDBT_Step* step){
+  NdbRestarter restarter;
+  int timeout = ctx->getProperty("TransactionDeadlockTimeout", TIMEOUT);
+  
+  NdbConfig conf(GETNDB(step)->getNodeId()+1);
+  unsigned int nodeId = conf.getMasterNodeId();
+  if (!conf.getProperty(nodeId,
+			NODE_TYPE_DB, 
+			CFG_DB_TRANSACTION_DEADLOCK_TIMEOUT,
+			&g_org_deadlock))
+    return NDBT_FAILED;
+  
+  g_err << "Setting timeout: " << timeout << endl;
+  int val[] = { DumpStateOrd::TcSetTransactionTimeout, timeout };
+  if(restarter.dumpStateAllNodes(val, 2) != 0){
+    return NDBT_FAILED;
+  }
+  
+  return NDBT_OK;
+}
+
+int
+getDeadlockTimeout(NDBT_Context* ctx, NDBT_Step* step){
+  NdbRestarter restarter;
+  
+  Uint32 val = 0;
+  NdbConfig conf(GETNDB(step)->getNodeId()+1);
+  unsigned int nodeId = conf.getMasterNodeId();
+  if (!conf.getProperty(nodeId,
+			NODE_TYPE_DB, 
+			CFG_DB_TRANSACTION_DEADLOCK_TIMEOUT,
+			&val))
+    return NDBT_FAILED;
+
+  if (val < 120000)
+    val = 120000;
+  ctx->setProperty("TransactionDeadlockTimeout", 4*val);
+  
+  return NDBT_OK;
+}
+
+int
+resetDeadlockTimeout(NDBT_Context* ctx, NDBT_Step* step){
+  NdbRestarter restarter;
+  
+  int val[] = { DumpStateOrd::TcSetTransactionTimeout, g_org_deadlock };
+  if(restarter.dumpStateAllNodes(val, 2) != 0){
+    return NDBT_FAILED;
+  }
+  
+  return NDBT_OK;
+}
+
 
 int runLoadTable(NDBT_Context* ctx, NDBT_Step* step){
 
@@ -333,6 +388,43 @@ int runBuddyTransNoTimeout(NDBT_Context* ctx, NDBT_Step* step){
   return result;
 }
 
+int 
+runError4012(NDBT_Context* ctx, NDBT_Step* step){
+  int result = NDBT_OK;
+  int loops = ctx->getNumLoops();
+  int stepNo = step->getStepNo();
+  
+  int timeout = ctx->getProperty("TransactionDeadlockTimeout", TIMEOUT);
+
+  HugoOperations hugoOps(*ctx->getTab());
+  Ndb* pNdb = GETNDB(step);
+
+  do{
+    // Commit transaction
+    CHECK(hugoOps.startTransaction(pNdb) == 0);
+    CHECK(hugoOps.pkUpdateRecord(pNdb, 0) == 0);
+    int ret = hugoOps.execute_NoCommit(pNdb);
+    if (ret == 0)
+    {
+      int sleep = timeout;
+      ndbout << "Sleeping for " << sleep << " milliseconds" << endl;
+      NdbSleep_MilliSleep(sleep);
+      
+      // Expect that transaction has NOT timed-out
+      CHECK(hugoOps.execute_Commit(pNdb) == 0);
+    }
+    else
+    {
+      CHECK(ret == 4012);
+    }
+  } while(false);
+  
+  hugoOps.closeTransaction(pNdb);
+  
+  return result;
+}
+
+
 NDBT_TESTSUITE(testTimeout);
 TESTCASE("DontTimeoutTransaction", 
 	 "Test that the transaction does not timeout "\
@@ -403,6 +495,15 @@ TESTCASE("BuddyTransNoTimeout5",
   FINALIZER(resetTransactionTimeout);
   FINALIZER(runClearTable);
 }
+TESTCASE("Error4012", ""){
+  TC_PROPERTY("TransactionDeadlockTimeout", 120000);
+  INITIALIZER(runLoadTable);
+  INITIALIZER(getDeadlockTimeout);
+  INITIALIZER(setDeadlockTimeout);
+  STEPS(runError4012, 2);
+  FINALIZER(runClearTable);
+}
+
 NDBT_TESTSUITE_END(testTimeout);
 
 int main(int argc, const char** argv){

storage/ndb/test/run-test/Makefile.am

@@ -7,14 +7,13 @@ include $(top_srcdir)/storage/ndb/config/type_mgmapiclient.mk.am
 
 test_PROGRAMS = atrt
 test_DATA=daily-basic-tests.txt daily-devel-tests.txt 16node-tests.txt \
-          conf-daily-basic-ndbmaster.txt \
-          conf-daily-basic-shark.txt \
-          conf-daily-devel-ndbmaster.txt \
-          conf-daily-sql-ndbmaster.txt \
-          conf-daily-basic-dl145a.txt \
           conf-daily-basic-ndb08.txt \
           conf-daily-devel-ndb08.txt \
           conf-daily-sql-ndb08.txt
+          conf-ndbmaster.txt \
+          conf-shark.txt \
+          conf-dl145a.txt
+
 test_SCRIPTS=atrt-analyze-result.sh atrt-gather-result.sh atrt-setup.sh \
           atrt-clear-result.sh make-config.sh make-index.sh make-html-reports.sh
 

storage/ndb/test/run-test/conf-daily-devel-ndbmaster.txt

@@ -17,3 +17,6 @@ FileSystemPath: CHOOSE_dir/run
 PortNumber: 16000
 ArbitrationRank: 1
 DataDir: .
+
+[TCP DEFAULT]
+SendBufferMemory: 2M

storage/ndb/test/run-test/conf-daily-basic-dl145a.txt → storage/ndb/test/run-test/conf-dl145a.txt

@@ -17,3 +17,6 @@ FileSystemPath: /home/ndbdev/autotest/run
 PortNumber: 14000
 ArbitrationRank: 1
 DataDir: .
+
+[TCP DEFAULT]
+SendBufferMemory: 2M

storage/ndb/test/run-test/conf-daily-basic-ndbmaster.txt → storage/ndb/test/run-test/conf-ndbmaster.txt

@@ -17,3 +17,6 @@ FileSystemPath: CHOOSE_dir/run
 PortNumber: 14000
 ArbitrationRank: 1
 DataDir: .
+
+[TCP DEFAULT]
+SendBufferMemory: 2M

storage/ndb/test/run-test/conf-daily-basic-shark.txt → storage/ndb/test/run-test/conf-shark.txt

@@ -17,3 +17,6 @@ FileSystemPath: /space/autotest/run
 PortNumber: 14000
 ArbitrationRank: 1
 DataDir: .
+
+[TCP DEFAULT]
+SendBufferMemory: 2M

storage/ndb/test/run-test/daily-basic-tests.txt

@@ -445,6 +445,18 @@ max-time: 500
 cmd: testNodeRestart
 args: -n Bug15685 T1
 
+max-time: 500
+cmd: testNodeRestart
+args: -n Bug16772 T1
+
+#max-time: 500
+#cmd: testSystemRestart
+#args: -n Bug18385 T1
+#
+max-time: 500
+cmd: testNodeRestart
+args: -n Bug18414 T1
+
 #
 # DICT TESTS
 max-time: 1500

storage/ndb/test/run-test/ndb-autotest.sh

@@ -13,7 +13,7 @@ save_args=$*
 VERSION="ndb-autotest.sh version 1.04"
 
 DATE=`date '+%Y-%m-%d'`
-HOST=`hostname`
+HOST=`hostname -s`
 export DATE HOST
 
 set -e
@@ -67,7 +67,7 @@ done
 
 if [ -f $conf ]
 then
-	. ./$conf
+	. $conf
 else
 	echo "Can't find config file: $conf"
 	exit
@@ -299,9 +299,12 @@ choose_conf(){
     elif [ -f $test_dir/conf-$1.txt ]
     then
 	echo "$test_dir/conf-$1.txt"
+    elif [ -f $test_dir/conf-$HOST.txt ]
+	echo "$test_dir/conf-$HOST.txt"
     else
 	echo "Unable to find conf file looked for" 1>&2
 	echo "$test_dir/conf-$1-$HOST.txt and" 1>&2
+	echo "$test_dir/conf-$HOST.txt" 1>&2
 	echo "$test_dir/conf-$1.txt" 1>&2
 	exit
     fi

storage/ndb/test/src/NdbRestarter.cpp

@@ -174,6 +174,39 @@ NdbRestarter::getRandomNodeOtherNodeGroup(int nodeId, int rand){
   return -1;
 }
 
+int
+NdbRestarter::getRandomNodeSameNodeGroup(int nodeId, int rand){
+  if (!isConnected())
+    return -1;
+  
+  if (getStatus() != 0)
+    return -1;
+  
+  int node_group = -1;
+  for(size_t i = 0; i < ndbNodes.size(); i++){
+    if(ndbNodes[i].node_id == nodeId){
+      node_group = ndbNodes[i].node_group;
+      break;
+    }
+  }
+  if(node_group == -1){
+    return -1;
+  }
+
+  Uint32 counter = 0;
+  rand = rand % ndbNodes.size();
+  while(counter++ < ndbNodes.size() && 
+	(ndbNodes[rand].node_id == nodeId || 
+	 ndbNodes[rand].node_group != node_group))
+    rand = (rand + 1) % ndbNodes.size();
+  
+  if(ndbNodes[rand].node_group == node_group &&
+     ndbNodes[rand].node_id != nodeId)
+    return ndbNodes[rand].node_id;
+  
+  return -1;
+}
+
 int 
 NdbRestarter::waitClusterStarted(unsigned int _timeout){
   return waitClusterState(NDB_MGM_NODE_STATUS_STARTED, _timeout);

storage/ndb/tools/desc.cpp

@@ -30,6 +30,7 @@ NDB_STD_OPTS_VARS;
 
 static const char* _dbname = "TEST_DB";
 static int _unqualified = 0;
+static int _partinfo = 0;
 static struct my_option my_long_options[] =
 {
   NDB_STD_OPTS("ndb_desc"),
@@ -39,6 +40,9 @@ static struct my_option my_long_options[] =
   { "unqualified", 'u', "Use unqualified table names",
     (gptr*) &_unqualified, (gptr*) &_unqualified, 0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 }, 
+  { "extra-partition-info", 'p', "Print more info per partition",
+    (gptr*) &_partinfo, (gptr*) &_partinfo, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 }, 
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 static void usage()
@@ -52,6 +56,8 @@ static void usage()
   my_print_variables(my_long_options);
 }
 
+static void print_part_info(Ndb* pNdb, NDBT_Table* pTab);
+
 int main(int argc, char** argv){
   NDB_INIT(argv[0]);
   const char *load_default_groups[]= { "mysql_cluster",0 };
@@ -260,5 +266,76 @@ int desc_table(Ndb *myndb, char* name)
   }
   ndbout << endl;
 
+  if (_partinfo)
+    print_part_info(myndb, pTab);
+	
   return 1;
 }
+
+struct InfoInfo
+{
+  const char * m_title;
+  NdbRecAttr* m_rec_attr;
+  const NdbDictionary::Column* m_column;
+};
+
+
+static 
+void print_part_info(Ndb* pNdb, NDBT_Table* pTab)
+{
+  InfoInfo g_part_info[] = {
+    { "Partition", 0, NdbDictionary::Column::FRAGMENT },
+    { "Row count", 0, NdbDictionary::Column::ROW_COUNT },
+    { "Commit count", 0, NdbDictionary::Column::COMMIT_COUNT },
+    { "Frag memory", 0, NdbDictionary::Column::FRAGMENT_MEMORY },
+    { 0, 0, 0 }
+  };
+
+  ndbout << "-- Per partition info -- " << endl;
+  
+  NdbConnection* pTrans = pNdb->startTransaction();
+  if (pTrans == 0)
+    return;
+  
+  do
+  {
+    NdbScanOperation* pOp= pTrans->getNdbScanOperation(pTab->getName());
+    if (pOp == NULL)
+      break;
+    
+    int rs = pOp->readTuples(NdbOperation::LM_CommittedRead); 
+    if (rs != 0)
+      break;
+    
+    if (pOp->interpret_exit_last_row() != 0)
+      break;
+    
+    Uint32 i = 0;
+    for(i = 0; g_part_info[i].m_title != 0; i++)
+    {
+      if ((g_part_info[i].m_rec_attr = pOp->getValue(g_part_info[i].m_column)) == 0)
+	break;
+    }
+
+    if (g_part_info[i].m_title != 0)
+      break;
+
+    if (pTrans->execute(NoCommit) != 0)
+      break;
+	
+    for (i = 0; g_part_info[i].m_title != 0; i++)
+      ndbout << g_part_info[i].m_title << "\t";
+    ndbout << endl;
+    
+    while(pOp->nextResult() == 0)
+    {
+      for(i = 0; g_part_info[i].m_title != 0; i++)
+      {
+	ndbout << *g_part_info[i].m_rec_attr << "\t";
+      }
+      ndbout << endl;
+    }
+  } while(0);
+  
+  pTrans->close();
+}