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

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

17 files changed, 582 insertions, 329 deletions

diff --git a/ndb/include/util/File.hpp b/ndb/include/util/File.hpp
index 3ed0ad7a6f9..52b00f575f4 100644
--- a/ndb/include/util/File.hpp
+++ b/ndb/include/util/File.hpp
@@ -29,6 +29,14 @@ class File_class
 {
 public:
   /**
+   * Returns time for last contents modification of a file.
+   *
+   * @param aFileName a filename to check.
+   * @return the time for last contents modificaton of the file.
+   */
+  static time_t mtime(const char* aFileName);      
+
+  /**
    * Returns true if the file exist.
    *
    * @param aFileName a filename to check.
diff --git a/ndb/include/util/ndb_opts.h b/ndb/include/util/ndb_opts.h
index 462d9996582..ac2a48d6fe9 100644
--- a/ndb/include/util/ndb_opts.h
+++ b/ndb/include/util/ndb_opts.h
@@ -49,7 +49,7 @@ my_bool opt_core;
   { "ndb-connectstring", OPT_NDB_CONNECTSTRING, \
     "Set connect string for connecting to ndb_mgmd. " \
     "Syntax: \"[nodeid=<id>;][host=]<hostname>[:<port>]\". " \
-    "Overides specifying entries in NDB_CONNECTSTRING and Ndb.cfg", \
+    "Overrides specifying entries in NDB_CONNECTSTRING and my.cnf", \
     (gptr*) &opt_connect_str, (gptr*) &opt_connect_str, 0, \
     GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },\
   { "ndb-shm", OPT_NDB_SHM,\
diff --git a/ndb/src/common/logger/FileLogHandler.cpp b/ndb/src/common/logger/FileLogHandler.cpp
index 3d29e63ac1f..b8859630406 100644
--- a/ndb/src/common/logger/FileLogHandler.cpp
+++ b/ndb/src/common/logger/FileLogHandler.cpp
@@ -147,6 +147,7 @@ FileLogHandler::createNewFile()
   bool rc = true;	
   int fileNo = 1;
   char newName[PATH_MAX];
+  time_t newMtime, preMtime = 0;
 
   do
   {
@@ -159,7 +160,15 @@ FileLogHandler::createNewFile()
     }		
     BaseString::snprintf(newName, sizeof(newName),
 	       "%s.%d", m_pLogFile->getName(), fileNo++); 
-    
+    newMtime = File_class::mtime(newName);
+    if (newMtime < preMtime) 
+    {
+      break;
+    }
+    else
+    {
+      preMtime = newMtime;
+    }
   } while (File_class::exists(newName));
   
   m_pLogFile->close();	
diff --git a/ndb/src/common/util/File.cpp b/ndb/src/common/util/File.cpp
index e514ad8e122..12626f29e7d 100644
--- a/ndb/src/common/util/File.cpp
+++ b/ndb/src/common/util/File.cpp
@@ -24,6 +24,18 @@
 //
 // PUBLIC
 //
+time_t 
+File_class::mtime(const char* aFileName)
+{
+  MY_STAT stmp;
+  time_t rc = 0;
+
+  if (my_stat(aFileName, &stmp, MYF(0)) != NULL) {
+    rc = stmp.st_mtime;
+  }
+
+  return rc;
+}
 
 bool 
 File_class::exists(const char* aFileName)
diff --git a/ndb/src/kernel/blocks/dbdict/Dbdict.cpp b/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
index 2bb429aeabc..2c6fffe851c 100644
--- a/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
+++ b/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
@@ -2313,7 +2313,8 @@ void Dbdict::checkSchemaStatus(Signal* signal)
 	tablePtr.p->tableType = (DictTabInfo::TableType)oldEntry->m_tableType;
 	
 	// On NR get index from master because index state is not on file
-	const bool file = c_systemRestart || tablePtr.p->isTable();
+	const bool file = (* newEntry == * oldEntry) &&
+	  (c_systemRestart || tablePtr.p->isTable());
 	restartCreateTab(signal, tableId, oldEntry, file);
 
 	return;
diff --git a/ndb/src/kernel/blocks/dbdih/Dbdih.hpp b/ndb/src/kernel/blocks/dbdih/Dbdih.hpp
index 78acf1ffd19..559d13f6e4b 100644
--- a/ndb/src/kernel/blocks/dbdih/Dbdih.hpp
+++ b/ndb/src/kernel/blocks/dbdih/Dbdih.hpp
@@ -1044,6 +1044,8 @@ private:
   void removeStoredReplica(FragmentstorePtr regFragptr,
                            ReplicaRecordPtr replicaPtr);
   void searchStoredReplicas(FragmentstorePtr regFragptr);
+  bool setup_create_replica(FragmentstorePtr, CreateReplicaRecord*,
+			    ConstPtr<ReplicaRecord>);
   void updateNodeInfo(FragmentstorePtr regFragptr);
 
 //------------------------------------
diff --git a/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp b/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
index 02ec5782c3e..7ae7db967b2 100644
--- a/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
+++ b/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
@@ -1265,9 +1265,9 @@ void Dbdih::execNDB_STTOR(Signal* signal)
       if (isMaster()) {
 	jam();
 	systemRestartTakeOverLab(signal);
-	if (anyActiveTakeOver() && false) {
+	if (anyActiveTakeOver())
+	{
 	  jam();
-	  ndbout_c("1 - anyActiveTakeOver == true");
 	  return;
 	}
       }
@@ -2260,6 +2260,8 @@ Dbdih::systemRestartTakeOverLab(Signal* signal)
 	// NOT ACTIVE NODES THAT HAVE NOT YET BEEN TAKEN OVER NEEDS TAKE OVER
 	// IMMEDIATELY. IF WE ARE ALIVE WE TAKE OVER OUR OWN NODE.
 	/*-------------------------------------------------------------------*/
+	infoEvent("Take over of node %d started", 
+		  nodePtr.i);
 	startTakeOver(signal, RNIL, nodePtr.i, nodePtr.i);
       }//if
       break;
@@ -2372,6 +2374,12 @@ void Dbdih::nodeRestartTakeOver(Signal* signal, Uint32 startNodeId)
      *--------------------------------------------------------------------*/
     Uint32 takeOverNode = Sysfile::getTakeOverNode(startNodeId, 
 						   SYSFILE->takeOver);
+    if(takeOverNode == 0){
+      jam();
+      warningEvent("Bug in take-over code restarting");
+      takeOverNode = startNodeId;
+    }
+
     startTakeOver(signal, RNIL, startNodeId, takeOverNode);
     break;
   }
@@ -2525,7 +2533,14 @@ void Dbdih::startTakeOver(Signal* signal,
   Sysfile::setTakeOverNode(takeOverPtr.p->toFailedNode, SYSFILE->takeOver,
 			   startNode);
   takeOverPtr.p->toMasterStatus = TakeOverRecord::TO_START_COPY;
-  
+
+  if (getNodeState().getSystemRestartInProgress())
+  {
+    jam();
+    checkToCopy();
+    checkToCopyCompleted(signal);
+    return;
+  }
   cstartGcpNow = true;
 }//Dbdih::startTakeOver()
 
@@ -3273,6 +3288,18 @@ void Dbdih::toCopyCompletedLab(Signal * signal, TakeOverRecordPtr takeOverPtr)
   signal->theData[1] = takeOverPtr.p->toStartingNode;
   sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
 
+  if (getNodeState().getSystemRestartInProgress())
+  {
+    jam();
+    infoEvent("Take over of node %d complete", takeOverPtr.p->toStartingNode);
+    setNodeActiveStatus(takeOverPtr.p->toStartingNode, Sysfile::NS_Active);
+    takeOverPtr.p->toMasterStatus = TakeOverRecord::WAIT_LCP;
+    takeOverCompleted(takeOverPtr.p->toStartingNode);
+    checkToCopy();
+    checkToCopyCompleted(signal);
+    return;
+  }
+  
   c_lcpState.immediateLcpStart = true;
   takeOverPtr.p->toMasterStatus = TakeOverRecord::WAIT_LCP;
   
@@ -3379,16 +3406,12 @@ void Dbdih::execEND_TOCONF(Signal* signal)
   }//if
   endTakeOver(takeOverPtr.i);
 
-  ndbout_c("2 - endTakeOver");
   if (cstartPhase == ZNDB_SPH4) {
     jam();
-    ndbrequire(false);
     if (anyActiveTakeOver()) {
       jam();
-      ndbout_c("4 - anyActiveTakeOver == true");
       return;
     }//if
-    ndbout_c("5 - anyActiveTakeOver == false -> ndbsttorry10Lab");
     ndbsttorry10Lab(signal, __LINE__);
     return;
   }//if
@@ -8321,14 +8344,30 @@ Dbdih::resetReplicaSr(TabRecordPtr tabPtr){
 
 	  resetReplicaLcp(replicaPtr.p, newestRestorableGCI);
 
-	  /* -----------------------------------------------------------------
-	   *   LINK THE REPLICA INTO THE STORED REPLICA LIST. WE WILL USE THIS
-	   *   NODE AS A STORED REPLICA.                                      
-	   *   WE MUST FIRST LINK IT OUT OF THE LIST OF OLD STORED REPLICAS.  
-	   * --------------------------------------------------------------- */
-	  removeOldStoredReplica(fragPtr, replicaPtr);
-	  linkStoredReplica(fragPtr, replicaPtr);
-
+	  /**
+	   * Make sure we can also find REDO for restoring replica...
+	   */
+	  {
+	    CreateReplicaRecord createReplica;
+	    ConstPtr<ReplicaRecord> constReplicaPtr;
+	    constReplicaPtr.i = replicaPtr.i;
+	    constReplicaPtr.p = replicaPtr.p;
+	    if (setup_create_replica(fragPtr,
+				     &createReplica, constReplicaPtr))
+	    {
+	      removeOldStoredReplica(fragPtr, replicaPtr);
+	      linkStoredReplica(fragPtr, replicaPtr);
+	    }
+	    else
+	    {
+	      infoEvent("Forcing take-over of node %d due to unsufficient REDO"
+			" for table %d fragment: %d",
+			nodePtr.i, tabPtr.i, i);
+	      
+	      setNodeActiveStatus(nodePtr.i, 
+				  Sysfile::NS_NotActive_NotTakenOver);
+	    }
+	  }
 	}
         default:
 	  jam();
@@ -9376,6 +9415,7 @@ void Dbdih::calculateKeepGciLab(Signal* signal, Uint32 tableId, Uint32 fragId)
   FragmentstorePtr fragPtr;
   getFragstore(tabPtr.p, fragId, fragPtr);
   checkKeepGci(tabPtr, fragId, fragPtr.p, fragPtr.p->storedReplicas);
+  checkKeepGci(tabPtr, fragId, fragPtr.p, fragPtr.p->oldStoredReplicas);
   fragId++;
   if (fragId >= tabPtr.p->totalfragments) {
     jam();
@@ -9561,73 +9601,84 @@ void Dbdih::startNextChkpt(Signal* signal)
       nodePtr.i = replicaPtr.p->procNode;
       ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
       
-      if (replicaPtr.p->lcpOngoingFlag &&
-          replicaPtr.p->lcpIdStarted < lcpId) {
-        jam();
-	//-------------------------------------------------------------------
-	// We have found a replica on a node that performs local checkpoint
-	// that is alive and that have not yet been started.
-	//-------------------------------------------------------------------
-
-        if (nodePtr.p->noOfStartedChkpt < 2) {
-          jam();
-	  /**
-	   * Send LCP_FRAG_ORD to LQH
-	   */
-	  
-	  /**
-	   * Mark the replica so with lcpIdStarted == true
-	   */
-          replicaPtr.p->lcpIdStarted = lcpId;
-
-          Uint32 i = nodePtr.p->noOfStartedChkpt;
-          nodePtr.p->startedChkpt[i].tableId = tabPtr.i;
-          nodePtr.p->startedChkpt[i].fragId = curr.fragmentId;
-          nodePtr.p->startedChkpt[i].replicaPtr = replicaPtr.i;
-          nodePtr.p->noOfStartedChkpt = i + 1;
-
-	  sendLCP_FRAG_ORD(signal, nodePtr.p->startedChkpt[i]);
-        } else if (nodePtr.p->noOfQueuedChkpt < 2) {
-          jam();
-	  /**
-	   * Put LCP_FRAG_ORD "in queue"
-	   */
-	  
-	  /**
-	   * Mark the replica so with lcpIdStarted == true
-	   */
-          replicaPtr.p->lcpIdStarted = lcpId;
+      if (c_lcpState.m_participatingLQH.get(nodePtr.i))
+      {
+	if (replicaPtr.p->lcpOngoingFlag &&
+	    replicaPtr.p->lcpIdStarted < lcpId) 
+	{
+	  jam();
+	  //-------------------------------------------------------------------
+	  // We have found a replica on a node that performs local checkpoint
+	  // that is alive and that have not yet been started.
+	  //-------------------------------------------------------------------
 	  
-          Uint32 i = nodePtr.p->noOfQueuedChkpt;
-          nodePtr.p->queuedChkpt[i].tableId = tabPtr.i;
-          nodePtr.p->queuedChkpt[i].fragId = curr.fragmentId;
-          nodePtr.p->queuedChkpt[i].replicaPtr = replicaPtr.i;
-          nodePtr.p->noOfQueuedChkpt = i + 1;
-        } else {
-          jam();
+	  if (nodePtr.p->noOfStartedChkpt < 2) 
+	  {
+	    jam();
+	    /**
+	     * Send LCP_FRAG_ORD to LQH
+	     */
+	    
+	    /**
+	     * Mark the replica so with lcpIdStarted == true
+	     */
+	    replicaPtr.p->lcpIdStarted = lcpId;
 
-	  if(save){
+	    Uint32 i = nodePtr.p->noOfStartedChkpt;
+	    nodePtr.p->startedChkpt[i].tableId = tabPtr.i;
+	    nodePtr.p->startedChkpt[i].fragId = curr.fragmentId;
+	    nodePtr.p->startedChkpt[i].replicaPtr = replicaPtr.i;
+	    nodePtr.p->noOfStartedChkpt = i + 1;
+	    
+	    sendLCP_FRAG_ORD(signal, nodePtr.p->startedChkpt[i]);
+	  } 
+	  else if (nodePtr.p->noOfQueuedChkpt < 2) 
+	  {
+	    jam();
 	    /**
-	     * Stop increasing value on first that was "full"
+	     * Put LCP_FRAG_ORD "in queue"
 	     */
-	    c_lcpState.currentFragment = curr;
-	    save = false;
-	  }
-	  
-	  busyNodes.set(nodePtr.i);
-	  if(busyNodes.count() == lcpNodes){
+	    
 	    /**
-	     * There were no possibility to start the local checkpoint 
-	     * and it was not possible to queue it up. In this case we 
-	     * stop the start of local checkpoints until the nodes with a 
-	     * backlog have performed more checkpoints. We will return and 
-	     * will not continue the process of starting any more checkpoints.
+	     * Mark the replica so with lcpIdStarted == true
 	     */
-	    return;
+	    replicaPtr.p->lcpIdStarted = lcpId;
+	    
+	    Uint32 i = nodePtr.p->noOfQueuedChkpt;
+	    nodePtr.p->queuedChkpt[i].tableId = tabPtr.i;
+	    nodePtr.p->queuedChkpt[i].fragId = curr.fragmentId;
+	    nodePtr.p->queuedChkpt[i].replicaPtr = replicaPtr.i;
+	    nodePtr.p->noOfQueuedChkpt = i + 1;
+	  } 
+	  else 
+	  {
+	    jam();
+	    
+	    if(save)
+	    {
+	      /**
+	       * Stop increasing value on first that was "full"
+	       */
+	      c_lcpState.currentFragment = curr;
+	      save = false;
+	    }
+	    
+	    busyNodes.set(nodePtr.i);
+	    if(busyNodes.count() == lcpNodes)
+	    {
+	      /**
+	       * There were no possibility to start the local checkpoint 
+	       * and it was not possible to queue it up. In this case we 
+	       * stop the start of local checkpoints until the nodes with a 
+	       * backlog have performed more checkpoints. We will return and 
+	       * will not continue the process of starting any more checkpoints.
+	       */
+	      return;
+	    }//if
 	  }//if
-	}//if
-      }
-    }//while
+	}
+      }//while
+    }
     curr.fragmentId++;
     if (curr.fragmentId >= tabPtr.p->totalfragments) {
       jam();
@@ -12247,16 +12298,75 @@ void Dbdih::removeTooNewCrashedReplicas(ReplicaRecordPtr rtnReplicaPtr)
 /*               CHECKPOINT WITHOUT NEEDING ANY EXTRA LOGGING FACILITIES.*/
 /*               A MAXIMUM OF FOUR NODES IS RETRIEVED.                   */
 /*************************************************************************/
+bool
+Dbdih::setup_create_replica(FragmentstorePtr fragPtr,
+			    CreateReplicaRecord* createReplicaPtrP,
+			    ConstPtr<ReplicaRecord> replicaPtr)
+{
+  createReplicaPtrP->dataNodeId = replicaPtr.p->procNode;
+  createReplicaPtrP->replicaRec = replicaPtr.i;
+
+  /* ----------------------------------------------------------------- */
+  /*   WE NEED TO SEARCH FOR A PROPER LOCAL CHECKPOINT TO USE FOR THE  */
+  /*   SYSTEM RESTART.                                                 */
+  /* ----------------------------------------------------------------- */
+  Uint32 startGci;
+  Uint32 startLcpNo;
+  Uint32 stopGci = SYSFILE->newestRestorableGCI;
+  bool result = findStartGci(replicaPtr,
+			     stopGci,
+			     startGci,
+			     startLcpNo);
+  if (!result) 
+  {
+    jam();
+    /* --------------------------------------------------------------- */
+    /* WE COULD NOT FIND ANY LOCAL CHECKPOINT. THE FRAGMENT THUS DO NOT*/
+    /* CONTAIN ANY VALID LOCAL CHECKPOINT. IT DOES HOWEVER CONTAIN A   */
+    /* VALID FRAGMENT LOG. THUS BY FIRST CREATING THE FRAGMENT AND THEN*/
+    /* EXECUTING THE FRAGMENT LOG WE CAN CREATE THE FRAGMENT AS        */
+    /* DESIRED. THIS SHOULD ONLY OCCUR AFTER CREATING A FRAGMENT.      */
+    /*                                                                 */
+    /* TO INDICATE THAT NO LOCAL CHECKPOINT IS TO BE USED WE SET THE   */
+    /* LOCAL CHECKPOINT TO ZNIL.                                       */
+    /* --------------------------------------------------------------- */
+    createReplicaPtrP->lcpNo = ZNIL;
+  } 
+  else 
+  {
+    jam();
+    /* --------------------------------------------------------------- */
+    /* WE FOUND A PROPER LOCAL CHECKPOINT TO RESTART FROM.             */
+    /* SET LOCAL CHECKPOINT ID AND LOCAL CHECKPOINT NUMBER.            */
+    /* --------------------------------------------------------------- */
+    createReplicaPtrP->lcpNo = startLcpNo;
+    arrGuard(startLcpNo, MAX_LCP_STORED);
+    createReplicaPtrP->createLcpId = replicaPtr.p->lcpId[startLcpNo];
+  }//if
+  
+  
+  /* ----------------------------------------------------------------- */
+  /*   WE HAVE EITHER FOUND A LOCAL CHECKPOINT OR WE ARE PLANNING TO   */
+  /*   EXECUTE THE LOG FROM THE INITIAL CREATION OF THE TABLE. IN BOTH */
+  /*   CASES WE NEED TO FIND A SET OF LOGS THAT CAN EXECUTE SUCH THAT  */
+  /*   WE RECOVER TO THE SYSTEM RESTART GLOBAL CHECKPOINT.             */
+  /* -_--------------------------------------------------------------- */
+  return findLogNodes(createReplicaPtrP, fragPtr, startGci, stopGci);
+}			    
+
 void Dbdih::searchStoredReplicas(FragmentstorePtr fragPtr) 
 {
   Uint32 nextReplicaPtrI;
-  ConstPtr<ReplicaRecord> replicaPtr;
+  Ptr<ReplicaRecord> replicaPtr;
 
   replicaPtr.i = fragPtr.p->storedReplicas;
   while (replicaPtr.i != RNIL) {
     jam();
     ptrCheckGuard(replicaPtr, creplicaFileSize, replicaRecord);
     nextReplicaPtrI = replicaPtr.p->nextReplica;
+    ConstPtr<ReplicaRecord> constReplicaPtr;
+    constReplicaPtr.i = replicaPtr.i;
+    constReplicaPtr.p = replicaPtr.p;
     NodeRecordPtr nodePtr;
     nodePtr.i = replicaPtr.p->procNode;
     ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
@@ -12276,69 +12386,13 @@ void Dbdih::searchStoredReplicas(FragmentstorePtr fragPtr)
 	createReplicaPtr.i = cnoOfCreateReplicas;
 	ptrCheckGuard(createReplicaPtr, 4, createReplicaRecord);
 	cnoOfCreateReplicas++;
-	createReplicaPtr.p->dataNodeId = replicaPtr.p->procNode;
-	createReplicaPtr.p->replicaRec = replicaPtr.i;
-	/* ----------------------------------------------------------------- */
-	/*   WE NEED TO SEARCH FOR A PROPER LOCAL CHECKPOINT TO USE FOR THE  */
-	/*   SYSTEM RESTART.                                                 */
-	/* ----------------------------------------------------------------- */
-	Uint32 startGci;
-	Uint32 startLcpNo;
-	Uint32 stopGci = SYSFILE->newestRestorableGCI;
-	bool result = findStartGci(replicaPtr,
-				   stopGci,
-				   startGci,
-				   startLcpNo);
-	if (!result) {
-	  jam();
-	  /* --------------------------------------------------------------- */
-	  /* WE COULD NOT FIND ANY LOCAL CHECKPOINT. THE FRAGMENT THUS DO NOT*/
-	  /* CONTAIN ANY VALID LOCAL CHECKPOINT. IT DOES HOWEVER CONTAIN A   */
-	  /* VALID FRAGMENT LOG. THUS BY FIRST CREATING THE FRAGMENT AND THEN*/
-	  /* EXECUTING THE FRAGMENT LOG WE CAN CREATE THE FRAGMENT AS        */
-	  /* DESIRED. THIS SHOULD ONLY OCCUR AFTER CREATING A FRAGMENT.      */
-	  /*                                                                 */
-	  /* TO INDICATE THAT NO LOCAL CHECKPOINT IS TO BE USED WE SET THE   */
-	  /* LOCAL CHECKPOINT TO ZNIL.                                       */
-	  /* --------------------------------------------------------------- */
-	  createReplicaPtr.p->lcpNo = ZNIL;
-	} else {
-	  jam();
-	  /* --------------------------------------------------------------- */
-	  /* WE FOUND A PROPER LOCAL CHECKPOINT TO RESTART FROM.             */
-	  /* SET LOCAL CHECKPOINT ID AND LOCAL CHECKPOINT NUMBER.            */
-	  /* --------------------------------------------------------------- */
-	  createReplicaPtr.p->lcpNo = startLcpNo;
-	  arrGuard(startLcpNo, MAX_LCP_STORED);
-	  createReplicaPtr.p->createLcpId = replicaPtr.p->lcpId[startLcpNo];
-	}//if
-
-	if(ERROR_INSERTED(7073) || ERROR_INSERTED(7074)){
-	  jam();
-	  nodePtr.p->nodeStatus = NodeRecord::DEAD;
-	}
-
-	/* ----------------------------------------------------------------- */
-	/*   WE HAVE EITHER FOUND A LOCAL CHECKPOINT OR WE ARE PLANNING TO   */
-	/*   EXECUTE THE LOG FROM THE INITIAL CREATION OF THE TABLE. IN BOTH */
-	/*   CASES WE NEED TO FIND A SET OF LOGS THAT CAN EXECUTE SUCH THAT  */
-	/*   WE RECOVER TO THE SYSTEM RESTART GLOBAL CHECKPOINT.             */
-	/* -_--------------------------------------------------------------- */
-	if (!findLogNodes(createReplicaPtr.p, fragPtr, startGci, stopGci)) {
-	  jam();
-	  /* --------------------------------------------------------------- */
-	  /* WE WERE NOT ABLE TO FIND ANY WAY OF RESTORING THIS REPLICA.     */
-	  /* THIS IS A POTENTIAL SYSTEM ERROR.                               */
-	  /* --------------------------------------------------------------- */
-	  cnoOfCreateReplicas--;
-	  return;
-	}//if
-	
-	if(ERROR_INSERTED(7073) || ERROR_INSERTED(7074)){
-	  jam();
-	  nodePtr.p->nodeStatus = NodeRecord::ALIVE;
-	}
 	
+	/**
+	 * Should have been checked in resetReplicaSr
+	 */
+	ndbrequire(setup_create_replica(fragPtr,
+					createReplicaPtr.p, 
+					constReplicaPtr));
 	break;
       }
       default:
diff --git a/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp b/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
index ab0981a98ef..59e6bd35baf 100644
--- a/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
+++ b/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
@@ -984,13 +984,6 @@ Dbtc::handleFailedApiNode(Signal* signal,
         TloopCount += 64;
         break;
       case CS_CONNECTED:
-        /*********************************************************************/
-        // The api record is connected to failed node. We need to release the 
-        // connection and set it in a disconnected state.
-        /*********************************************************************/
-        jam();
-        releaseApiCon(signal, apiConnectptr.i);
-        break;
       case CS_REC_COMMITTING:
       case CS_RECEIVING:
       case CS_STARTED:
diff --git a/ndb/src/mgmclient/CommandInterpreter.cpp b/ndb/src/mgmclient/CommandInterpreter.cpp
index 72debcc26a9..65d5b038707 100644
--- a/ndb/src/mgmclient/CommandInterpreter.cpp
+++ b/ndb/src/mgmclient/CommandInterpreter.cpp
@@ -67,8 +67,9 @@ private:
    *                                command will be sent to all DB processes.
    *   @param  allAfterFirstToken:  What the client gave after the 
    *                                first token on the command line
+   *   @return: 0 if analyseAfterFirstToken succeeds, otherwise -1 
    */
-  void analyseAfterFirstToken(int processId, char* allAfterFirstTokenCstr);
+  int  analyseAfterFirstToken(int processId, char* allAfterFirstTokenCstr);
 
   /**
    *   Parse the block specification part of the LOG* commands,
@@ -93,38 +94,38 @@ private:
    *   this case "22". Each function is responsible to check the parameters
    *   argument.
    */
-  void executeHelp(char* parameters);
-  void executeShow(char* parameters);
-  void executeConnect(char* parameters);
-  void executePurge(char* parameters);
+  int  executeHelp(char* parameters);
+  int  executeShow(char* parameters);
+  int  executeConnect(char* parameters);
+  int  executePurge(char* parameters);
   int  executeShutdown(char* parameters);
   void executeRun(char* parameters);
   void executeInfo(char* parameters);
   void executeClusterLog(char* parameters);
 
 public:
-  void executeStop(int processId, const char* parameters, bool all);
-  void executeEnterSingleUser(char* parameters);
-  void executeExitSingleUser(char* parameters);
-  void executeStart(int processId, const char* parameters, bool all);
-  void executeRestart(int processId, const char* parameters, bool all);
-  void executeLogLevel(int processId, const char* parameters, bool all);
-  void executeError(int processId, const char* parameters, bool all);
-  void executeLog(int processId, const char* parameters, bool all);
-  void executeLogIn(int processId, const char* parameters, bool all);
-  void executeLogOut(int processId, const char* parameters, bool all);
-  void executeLogOff(int processId, const char* parameters, bool all);
-  void executeTestOn(int processId, const char* parameters, bool all);
-  void executeTestOff(int processId, const char* parameters, bool all);
-  void executeSet(int processId, const char* parameters, bool all);
-  void executeGetStat(int processId, const char* parameters, bool all);
-  void executeStatus(int processId, const char* parameters, bool all);
-  void executeEventReporting(int processId, const char* parameters, bool all);
-  void executeDumpState(int processId, const char* parameters, bool all);
-  int executeStartBackup(char * parameters);
-  void executeAbortBackup(char * parameters);
-
-  void executeRep(char* parameters);
+  int  executeStop(int processId, const char* parameters, bool all);
+  int  executeEnterSingleUser(char* parameters);
+  int  executeExitSingleUser(char* parameters);
+  int  executeStart(int processId, const char* parameters, bool all);
+  int  executeRestart(int processId, const char* parameters, bool all);
+  int  executeLogLevel(int processId, const char* parameters, bool all);
+  int  executeError(int processId, const char* parameters, bool all);
+  int  executeLog(int processId, const char* parameters, bool all);
+  int  executeLogIn(int processId, const char* parameters, bool all);
+  int  executeLogOut(int processId, const char* parameters, bool all);
+  int  executeLogOff(int processId, const char* parameters, bool all);
+  int  executeTestOn(int processId, const char* parameters, bool all);
+  int  executeTestOff(int processId, const char* parameters, bool all);
+  int  executeSet(int processId, const char* parameters, bool all);
+  int  executeGetStat(int processId, const char* parameters, bool all);
+  int  executeStatus(int processId, const char* parameters, bool all);
+  int  executeEventReporting(int processId, const char* parameters, bool all);
+  int  executeDumpState(int processId, const char* parameters, bool all);
+  int  executeStartBackup(char * parameters);
+  int  executeAbortBackup(char * parameters);
+
+  int  executeRep(char* parameters);
 
   void executeCpc(char * parameters);
 
@@ -136,7 +137,7 @@ public:
    * A execute function definition
    */
 public:
-  typedef void (CommandInterpreter::* ExecuteFunction)(int processId, 
+  typedef int (CommandInterpreter::* ExecuteFunction)(int processId, 
 						       const char * param, 
 						       bool all);
   
@@ -148,7 +149,7 @@ private:
   /**
    * 
    */
-  void executeForAll(const char * cmd, 
+  int  executeForAll(const char * cmd, 
 		     ExecuteFunction fun,
 		     const char * param);
 
@@ -606,6 +607,7 @@ CommandInterpreter::execute_impl(const char *_line)
 
   char * line;
   if(_line == NULL) {
+    m_error = -1;
     DBUG_RETURN(false);
   }
   line = my_strdup(_line,MYF(MY_WME));
@@ -636,16 +638,17 @@ CommandInterpreter::execute_impl(const char *_line)
   
   if (strcasecmp(firstToken, "HELP") == 0 ||
       strcasecmp(firstToken, "?") == 0) {
-    executeHelp(allAfterFirstToken);
+    m_error = executeHelp(allAfterFirstToken);
     DBUG_RETURN(true);
   }
   else if (strcasecmp(firstToken, "CONNECT") == 0) {
-    executeConnect(allAfterFirstToken);
+    m_error = executeConnect(allAfterFirstToken);
     DBUG_RETURN(true);
   }
   else if (strcasecmp(firstToken, "SLEEP") == 0) {
     if (allAfterFirstToken)
-      sleep(atoi(allAfterFirstToken));
+      if (sleep(atoi(allAfterFirstToken)) != 0 )
+        m_error = -1;
     DBUG_RETURN(true);
   }
   else if((strcasecmp(firstToken, "QUIT") == 0 ||
@@ -655,11 +658,13 @@ CommandInterpreter::execute_impl(const char *_line)
     DBUG_RETURN(false);
   }
 
-  if (!connect())
+  if (!connect()){
+    m_error = -1;
     DBUG_RETURN(true);
+  }
 
   if (strcasecmp(firstToken, "SHOW") == 0) {
-    executeShow(allAfterFirstToken);
+    m_error = executeShow(allAfterFirstToken);
     DBUG_RETURN(true);
   }
   else if (strcasecmp(firstToken, "SHUTDOWN") == 0) {
@@ -679,17 +684,17 @@ CommandInterpreter::execute_impl(const char *_line)
   else if(strcasecmp(firstToken, "ABORT") == 0 &&
 	  allAfterFirstToken != NULL &&
 	  strncasecmp(allAfterFirstToken, "BACKUP", sizeof("BACKUP") - 1) == 0){
-    executeAbortBackup(allAfterFirstToken);
+    m_error = executeAbortBackup(allAfterFirstToken);
     DBUG_RETURN(true);
   }
   else if (strcasecmp(firstToken, "PURGE") == 0) {
-    executePurge(allAfterFirstToken);
+    m_error = executePurge(allAfterFirstToken);
     DBUG_RETURN(true);
   } 
 #ifdef HAVE_GLOBAL_REPLICATION
   else if(strcasecmp(firstToken, "REPLICATION") == 0 ||
 	  strcasecmp(firstToken, "REP") == 0) {
-    executeRep(allAfterFirstToken);
+    m_error = executeRep(allAfterFirstToken);
     DBUG_RETURN(true);
   }
 #endif // HAVE_GLOBAL_REPLICATION
@@ -697,18 +702,18 @@ CommandInterpreter::execute_impl(const char *_line)
 	  allAfterFirstToken != NULL &&
 	  strncasecmp(allAfterFirstToken, "SINGLE USER MODE ", 
 		  sizeof("SINGLE USER MODE") - 1) == 0){
-    executeEnterSingleUser(allAfterFirstToken);
+    m_error = executeEnterSingleUser(allAfterFirstToken);
     DBUG_RETURN(true);
   }
   else if(strcasecmp(firstToken, "EXIT") == 0 &&
 	  allAfterFirstToken != NULL &&
 	  strncasecmp(allAfterFirstToken, "SINGLE USER MODE ", 
 		  sizeof("SINGLE USER MODE") - 1) == 0){
-    executeExitSingleUser(allAfterFirstToken);
+    m_error = executeExitSingleUser(allAfterFirstToken);
     DBUG_RETURN(true);
   }
   else if (strcasecmp(firstToken, "ALL") == 0) {
-    analyseAfterFirstToken(-1, allAfterFirstToken);
+    m_error = analyseAfterFirstToken(-1, allAfterFirstToken);
   } else {
     /**
      * First token should be a digit, node ID
@@ -717,15 +722,17 @@ CommandInterpreter::execute_impl(const char *_line)
 
     if (! convert(firstToken, nodeId)) {
       invalid_command(_line);
+      m_error = -1;
       DBUG_RETURN(true);
     }
 
     if (nodeId <= 0) {
       ndbout << "Invalid node ID: " << firstToken << "." << endl;
+      m_error = -1;
       DBUG_RETURN(true);
     }
     
-    analyseAfterFirstToken(nodeId, allAfterFirstToken);
+    m_error = analyseAfterFirstToken(nodeId, allAfterFirstToken);
     
   }
   DBUG_RETURN(true);
@@ -759,14 +766,15 @@ static const CommandInterpreter::CommandFunctionPair commands[] = {
 
 //*****************************************************************************
 //*****************************************************************************
-void
+int
 CommandInterpreter::analyseAfterFirstToken(int processId,
 					   char* allAfterFirstToken) {
   
+  int retval = 0;
   if (emptyString(allAfterFirstToken)) {
     ndbout << "Expected a command after "
 	   << ((processId == -1) ? "ALL." : "node ID.") << endl;
-    return;
+    return -1;
   }
   
   char* secondToken = strtok(allAfterFirstToken, " ");
@@ -785,15 +793,16 @@ CommandInterpreter::analyseAfterFirstToken(int processId,
   
   if(fun == 0){
     invalid_command(secondToken);
-    return;
+    return -1;
   }
   
   if(processId == -1){
-    executeForAll(command, fun, allAfterSecondToken);
+    retval = executeForAll(command, fun, allAfterSecondToken);
   } else {
-    (this->*fun)(processId, allAfterSecondToken, false);
+    retval = (this->*fun)(processId, allAfterSecondToken, false);
   }
   ndbout << endl;
+  return retval;
 }
 
 /**
@@ -834,18 +843,20 @@ get_next_nodeid(struct ndb_mgm_cluster_state *cl,
   return 0;
 }
 
-void
+int
 CommandInterpreter::executeForAll(const char * cmd, ExecuteFunction fun, 
 				  const char * allAfterSecondToken)
 {
   int nodeId = 0;
+  int retval = 0;
+
   if(strcasecmp(cmd, "STOP") == 0) {
     ndbout_c("Executing STOP on all nodes.");
-    (this->*fun)(nodeId, allAfterSecondToken, true);
+    retval = (this->*fun)(nodeId, allAfterSecondToken, true);
   } else if(strcasecmp(cmd, "RESTART") == 0) {
     ndbout_c("Executing RESTART on all nodes.");
     ndbout_c("Starting shutdown. This may take a while. Please wait...");
-    (this->*fun)(nodeId, allAfterSecondToken, true);
+    retval = (this->*fun)(nodeId, allAfterSecondToken, true);
     ndbout_c("Trying to start all nodes of system.");
     ndbout_c("Use ALL STATUS to see the system start-up phases.");
   } else {
@@ -853,12 +864,13 @@ CommandInterpreter::executeForAll(const char * cmd, ExecuteFunction fun,
     if(cl == 0){
       ndbout_c("Unable get status from management server");
       printError();
-      return;
+      return -1;
     }
     NdbAutoPtr<char> ap1((char*)cl);
     while(get_next_nodeid(cl, &nodeId, NDB_MGM_NODE_TYPE_NDB))
-      (this->*fun)(nodeId, allAfterSecondToken, true);
+      retval = (this->*fun)(nodeId, allAfterSecondToken, true);
   }
+  return retval;
 }
 
 //*****************************************************************************
@@ -928,7 +940,7 @@ CommandInterpreter::parseBlockSpecification(const char* allAfterLog,
 /*****************************************************************************
  * HELP
  *****************************************************************************/
-void 
+int
 CommandInterpreter::executeHelp(char* parameters)
 {
   if (emptyString(parameters)) {
@@ -966,7 +978,9 @@ CommandInterpreter::executeHelp(char* parameters)
 #endif
   } else {
     invalid_command(parameters);
+    return -1;
   }
+  return 0;
 }
 
 
@@ -1110,7 +1124,7 @@ print_nodes(ndb_mgm_cluster_state *state, ndb_mgm_configuration_iterator *it,
   ndbout << endl;
 }
 
-void
+int
 CommandInterpreter::executePurge(char* parameters) 
 { 
   int command_ok= 0;
@@ -1129,7 +1143,7 @@ CommandInterpreter::executePurge(char* parameters)
 
   if (!command_ok) {
     ndbout_c("Unexpected command, expected: PURGE STALE SESSIONS");
-    return;
+    return -1;
   }
 
   int i;
@@ -1137,7 +1151,7 @@ CommandInterpreter::executePurge(char* parameters)
   
   if (ndb_mgm_purge_stale_sessions(m_mgmsrv, &str)) {
     ndbout_c("Command failed");
-    return;
+    return -1;
   }
   if (str) {
     ndbout_c("Purged sessions with node id's: %s", str);
@@ -1147,9 +1161,10 @@ CommandInterpreter::executePurge(char* parameters)
   {
     ndbout_c("No sessions purged");
   }
+  return 0;
 }
 
-void
+int
 CommandInterpreter::executeShow(char* parameters) 
 { 
   int i;
@@ -1158,7 +1173,7 @@ CommandInterpreter::executeShow(char* parameters)
     if(state == NULL) {
       ndbout_c("Could not get status");
       printError();
-      return;
+      return -1;
     }
     NdbAutoPtr<char> ap1((char*)state);
 
@@ -1166,7 +1181,7 @@ CommandInterpreter::executeShow(char* parameters)
     if(conf == 0){
       ndbout_c("Could not get configuration");
       printError();
-      return;
+      return -1;
     }
 
     ndb_mgm_configuration_iterator * it;
@@ -1174,7 +1189,7 @@ CommandInterpreter::executeShow(char* parameters)
 
     if(it == 0){
       ndbout_c("Unable to create config iterator");
-      return;
+      return -1;
     }
     NdbAutoPtr<ndb_mgm_configuration_iterator> ptr(it);
 
@@ -1208,7 +1223,7 @@ CommandInterpreter::executeShow(char* parameters)
 	break;
       case NDB_MGM_NODE_TYPE_UNKNOWN:
         ndbout << "Error: Unknown Node Type" << endl;
-        return;
+        return -1;
       case NDB_MGM_NODE_TYPE_REP:
 	abort();
       }
@@ -1220,7 +1235,7 @@ CommandInterpreter::executeShow(char* parameters)
     print_nodes(state, it, "ndb_mgmd", mgm_nodes, NDB_MGM_NODE_TYPE_MGM, 0);
     print_nodes(state, it, "mysqld",   api_nodes, NDB_MGM_NODE_TYPE_API, 0);
     //    ndbout << helpTextShow;
-    return;
+    return 0;
   } else if (strcasecmp(parameters, "PROPERTIES") == 0 ||
 	     strcasecmp(parameters, "PROP") == 0) {
     ndbout << "SHOW PROPERTIES is not yet implemented." << endl;
@@ -1237,22 +1252,28 @@ CommandInterpreter::executeShow(char* parameters)
     //           << endl; /* XXX */
   } else {
     ndbout << "Invalid argument." << endl;
+    return -1;
   }
+  return 0;
 }
 
-void
+int
 CommandInterpreter::executeConnect(char* parameters) 
 {
+  int retval;
   disconnect();
   if (!emptyString(parameters)) {
-    if (ndb_mgm_set_connectstring(m_mgmsrv,
+    if (retval = ndb_mgm_set_connectstring(m_mgmsrv,
 				  BaseString(parameters).trim().c_str()))
     {
       printError();
-      return;
+      return retval;
     }
   }
-  connect();
+  if ( connect() == false ){
+    return -1;
+  }
+  return 0;
 }
 
 //*****************************************************************************
@@ -1265,6 +1286,7 @@ CommandInterpreter::executeClusterLog(char* parameters)
   if (emptyString(parameters))
   {
     ndbout << "Missing argument." << endl;
+    m_error = -1;
     DBUG_VOID_RETURN;
   }
 
@@ -1280,6 +1302,7 @@ CommandInterpreter::executeClusterLog(char* parameters)
   if(enabled == NULL) {
     ndbout << "Couldn't get status" << endl;
     printError();
+    m_error = -1;
     DBUG_VOID_RETURN;
   }
 
@@ -1291,6 +1314,7 @@ CommandInterpreter::executeClusterLog(char* parameters)
     if(enabled[0] == 0)
     {
       ndbout << "Cluster logging is disabled." << endl;
+      m_error = 0;
       DBUG_VOID_RETURN;
     }
 #if 0 
@@ -1309,6 +1333,7 @@ CommandInterpreter::executeClusterLog(char* parameters)
 	ndbout << BaseString(str).ndb_toupper() << " ";
     }
     ndbout << endl;
+    m_error = 0;
     DBUG_VOID_RETURN;
 
   } 
@@ -1327,6 +1352,7 @@ CommandInterpreter::executeClusterLog(char* parameters)
     enable= 1;
   } else {
     ndbout << "Invalid argument." << endl;
+    m_error = -1;
     DBUG_VOID_RETURN;
   }
 
@@ -1339,9 +1365,11 @@ CommandInterpreter::executeClusterLog(char* parameters)
     {
       ndbout << "Couldn't set filter" << endl;
       printError();
+      m_error = -1;
       DBUG_VOID_RETURN;
     }
     ndbout << "Cluster logging is " << (res_enable ? "enabled.":"disabled") << endl;
+    m_error = 0;
     DBUG_VOID_RETURN;
   }
 
@@ -1368,6 +1396,7 @@ CommandInterpreter::executeClusterLog(char* parameters)
     }
     if (severity == NDB_MGM_ILLEGAL_CLUSTERLOG_LEVEL) {
       ndbout << "Invalid severity level: " << item << endl;
+      m_error = -1;
       DBUG_VOID_RETURN;
     }
 
@@ -1376,23 +1405,26 @@ CommandInterpreter::executeClusterLog(char* parameters)
     {
       ndbout << "Couldn't set filter" << endl;
       printError();
+      m_error = -1;
       DBUG_VOID_RETURN;
     }
     ndbout << BaseString(item).ndb_toupper().c_str() << " " << (res_enable ? "enabled":"disabled") << endl;
 
     item = strtok_r(NULL, " ", &tmpPtr);	
   } while(item != NULL);
-
+  
+  m_error = 0;
   DBUG_VOID_RETURN;
 } 
 
 //*****************************************************************************
 //*****************************************************************************
 
-void
+int
 CommandInterpreter::executeStop(int processId, const char *, bool all) 
 {
   int result = 0;
+  int retval = 0;
   if(all) {
     result = ndb_mgm_stop(m_mgmsrv, 0, 0);
   } else {
@@ -1401,6 +1433,7 @@ CommandInterpreter::executeStop(int processId, const char *, bool all)
   if (result < 0) {
     ndbout << "Shutdown failed." << endl;
     printError();
+    retval = -1;
   } else
     {
       if(all)
@@ -1408,9 +1441,10 @@ CommandInterpreter::executeStop(int processId, const char *, bool all)
       else
 	ndbout << "Node " << processId << " has shutdown." << endl;
     }
+  return retval;
 }
 
-void
+int
 CommandInterpreter::executeEnterSingleUser(char* parameters) 
 {
   strtok(parameters, " ");
@@ -1422,37 +1456,42 @@ CommandInterpreter::executeEnterSingleUser(char* parameters)
   if(id == 0 || sscanf(id, "%d", &nodeId) != 1){
     ndbout_c("Invalid arguments: expected <NodeId>");
     ndbout_c("Use SHOW to see what API nodes are configured");
-    return;
+    return -1;
   }
   int result = ndb_mgm_enter_single_user(m_mgmsrv, nodeId, &reply);
   
   if (result != 0) {
     ndbout_c("Entering single user mode for node %d failed", nodeId);
     printError();
+    return -1;
   } else {
     ndbout_c("Single user mode entered");
     ndbout_c("Access is granted for API node %d only.", nodeId);
   }
+  return 0;
 }
 
-void 
+int
 CommandInterpreter::executeExitSingleUser(char* parameters) 
 {
   int result = ndb_mgm_exit_single_user(m_mgmsrv, 0);
   if (result != 0) {
     ndbout_c("Exiting single user mode failed.");
     printError();
+    return -1;
   } else {
     ndbout_c("Exiting single user mode in progress.");
     ndbout_c("Use ALL STATUS or SHOW to see when single user mode has been exited.");
+    return 0;
   }
 }
 
-void
+int
 CommandInterpreter::executeStart(int processId, const char* parameters,
 				 bool all) 
 {
   int result;
+  int retval = 0;
   if(all) {
     result = ndb_mgm_start(m_mgmsrv, 0, 0);
   } else {
@@ -1462,6 +1501,7 @@ CommandInterpreter::executeStart(int processId, const char* parameters,
   if (result <= 0) {
     ndbout << "Start failed." << endl;
     printError();
+    retval = -1;
   } else
     {
       if(all)
@@ -1469,9 +1509,10 @@ CommandInterpreter::executeStart(int processId, const char* parameters,
       else
 	ndbout_c("Database node %d is being started.", processId);
     }
+  return retval;
 }
 
-void
+int
 CommandInterpreter::executeRestart(int processId, const char* parameters,
 				   bool all) 
 {
@@ -1479,6 +1520,7 @@ CommandInterpreter::executeRestart(int processId, const char* parameters,
   int nostart = 0;
   int initialstart = 0;
   int abort = 0;
+  int retval = 0;
 
   if(parameters != 0 && strlen(parameters) != 0){
     char * tmpString = my_strdup(parameters,MYF(MY_WME));
@@ -1507,6 +1549,7 @@ CommandInterpreter::executeRestart(int processId, const char* parameters,
   if (result <= 0) {
     ndbout.println("Restart failed.", result);
     printError();
+    retval = -1;
   } else
     {
       if(all)
@@ -1514,15 +1557,16 @@ CommandInterpreter::executeRestart(int processId, const char* parameters,
       else
 	ndbout_c("Node %d is being restarted.", processId);
     }
+  return retval;
 }
 
-void
+int
 CommandInterpreter::executeDumpState(int processId, const char* parameters,
 				     bool all) 
 {
   if(emptyString(parameters)){
     ndbout << "Expected argument" << endl;
-    return;
+    return -1;
   }
 
   Uint32 no = 0;
@@ -1539,7 +1583,7 @@ CommandInterpreter::executeDumpState(int processId, const char* parameters,
       ndbout << "Illegal value in argument to signal." << endl
 	     << "(Value must be between 0 and 0xffffffff.)" 
 	     << endl;
-      return;
+      return -1;
     }
     no++;
     item = strtok_r(NULL, " ", &tmpPtr);
@@ -1551,16 +1595,16 @@ CommandInterpreter::executeDumpState(int processId, const char* parameters,
   }
   
   struct ndb_mgm_reply reply;
-  ndb_mgm_dump_state(m_mgmsrv, processId, pars, no, &reply);
+  return ndb_mgm_dump_state(m_mgmsrv, processId, pars, no, &reply);
 }
 
-void 
+int
 CommandInterpreter::executeStatus(int processId, 
 				  const char* parameters, bool all) 
 {
   if (! emptyString(parameters)) {
     ndbout_c("No parameters expected to this command.");
-    return;
+    return -1;
   }
 
   ndb_mgm_node_status status;
@@ -1572,7 +1616,7 @@ CommandInterpreter::executeStatus(int processId,
   if(cl == NULL) {
     ndbout_c("Cannot get status of node %d.", processId);
     printError();
-    return;
+    return -1;
   }
   NdbAutoPtr<char> ap1((char*)cl);
 
@@ -1581,7 +1625,7 @@ CommandInterpreter::executeStatus(int processId,
     i++;
   if(cl->node_states[i].node_id != processId) {
     ndbout << processId << ": Node not found" << endl;
-    return;
+    return -1;
   }
   status = cl->node_states[i].node_status;
   startPhase = cl->node_states[i].start_phase;
@@ -1605,27 +1649,29 @@ CommandInterpreter::executeStatus(int processId,
 	     getBuild(version));
   else
     ndbout << endl;
+  
+  return 0;
 }
 
 
 //*****************************************************************************
 //*****************************************************************************
 
-void 
+int
 CommandInterpreter::executeLogLevel(int processId, const char* parameters, 
 				    bool all) 
 {
   (void) all;
   if (emptyString(parameters)) {
     ndbout << "Expected argument" << endl;
-    return;
+    return -1;
   } 
   BaseString tmp(parameters);
   Vector<BaseString> spec;
   tmp.split(spec, "=");
   if(spec.size() != 2){
     ndbout << "Invalid loglevel specification: " << parameters << endl;
-    return;
+    return -1;
   }
 
   spec[0].trim().ndb_toupper();
@@ -1635,14 +1681,14 @@ CommandInterpreter::executeLogLevel(int processId, const char* parameters,
     if(category < NDB_MGM_MIN_EVENT_CATEGORY ||
        category > NDB_MGM_MAX_EVENT_CATEGORY){
       ndbout << "Unknown category: \"" << spec[0].c_str() << "\"" << endl;
-      return;
+      return -1;
     }
   }
   
   int level = atoi(spec[1].c_str());
   if(level < 0 || level > 15){
     ndbout << "Invalid level: " << spec[1].c_str() << endl;
-    return;
+    return -1;
   }
   
   ndbout << "Executing LOGLEVEL on node " << processId << flush;
@@ -1658,20 +1704,22 @@ CommandInterpreter::executeLogLevel(int processId, const char* parameters,
   if (result < 0) {
     ndbout_c(" failed.");
     printError();
+    return -1;
   } else {
     ndbout_c(" OK!");
   }  
-  
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
-void CommandInterpreter::executeError(int processId, 
+int CommandInterpreter::executeError(int processId, 
 				      const char* parameters, bool /* all */) 
 {
+  int retval = 0;
   if (emptyString(parameters)) {
     ndbout << "Missing error number." << endl;
-    return;
+    return -1;
   }
 
   // Copy parameters since strtok will modify it
@@ -1682,29 +1730,30 @@ void CommandInterpreter::executeError(int processId,
   int errorNo;
   if (! convert(firstParameter, errorNo)) {
     ndbout << "Expected an integer." << endl;
-    return;
+    return -1;
   }
 
   char* allAfterFirstParameter = strtok(NULL, "\0");
   if (! emptyString(allAfterFirstParameter)) {
     ndbout << "Nothing expected after error number." << endl;
-    return;
+    return -1;
   }
 
-  ndb_mgm_insert_error(m_mgmsrv, processId, errorNo, NULL);
+  retval = ndb_mgm_insert_error(m_mgmsrv, processId, errorNo, NULL);
+  return retval;
 }
 
 //*****************************************************************************
 //*****************************************************************************
 
-void 
+int
 CommandInterpreter::executeLog(int processId,
 			       const char* parameters, bool all) 
 {
   struct ndb_mgm_reply reply;
   Vector<const char *> blocks;
   if (! parseBlockSpecification(parameters, blocks)) {
-    return;
+    return -1;
   }
   int len=1;
   Uint32 i;
@@ -1728,82 +1777,91 @@ CommandInterpreter::executeLog(int processId,
   if (result != 0) {
     ndbout_c("Execute LOG on node %d failed.", processId);
     printError();
+    return -1;
   }
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
-void 
+int
 CommandInterpreter::executeLogIn(int /* processId */,
 				 const char* parameters, bool /* all */) 
 {
   ndbout << "Command LOGIN not implemented." << endl;
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
-void 
+int
 CommandInterpreter::executeLogOut(int /*processId*/, 
 				  const char* parameters, bool /*all*/) 
 {
   ndbout << "Command LOGOUT not implemented." << endl;
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
-void 
+int
 CommandInterpreter::executeLogOff(int /*processId*/,
 				  const char* parameters, bool /*all*/) 
 {
   ndbout << "Command LOGOFF not implemented." << endl;
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
-void 
+int
 CommandInterpreter::executeTestOn(int processId,
 				  const char* parameters, bool /*all*/) 
 {
   if (! emptyString(parameters)) {
     ndbout << "No parameters expected to this command." << endl;
-    return;
+    return -1;
   }
   struct ndb_mgm_reply reply;
   int result = ndb_mgm_start_signallog(m_mgmsrv, processId, &reply);
   if (result != 0) {
     ndbout_c("Execute TESTON failed.");
     printError();
+    return -1;
   }
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
-void 
+int
 CommandInterpreter::executeTestOff(int processId,
 				   const char* parameters, bool /*all*/) 
 {
   if (! emptyString(parameters)) {
     ndbout << "No parameters expected to this command." << endl;
-    return;
+    return -1;
   }
   struct ndb_mgm_reply reply;
   int result = ndb_mgm_stop_signallog(m_mgmsrv, processId, &reply);
   if (result != 0) {
     ndbout_c("Execute TESTOFF failed.");
     printError();
+    return -1;
   }
+  return 0;
 }
 
 
 //*****************************************************************************
 //*****************************************************************************
-void 
+int 
 CommandInterpreter::executeSet(int /*processId*/, 
 			       const char* parameters, bool /*all*/) 
 {
   if (emptyString(parameters)) {
     ndbout << "Missing parameter name." << endl;
-    return;
+    return -1;
   }
 #if 0
   // Copy parameters since strtok will modify it
@@ -1867,17 +1925,18 @@ CommandInterpreter::executeSet(int /*processId*/,
       abort();
     }
   }
-#endif
+#endif 
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
-void CommandInterpreter::executeGetStat(int /*processId*/,
+int CommandInterpreter::executeGetStat(int /*processId*/,
 					const char* parameters, bool /*all*/) 
 {
   if (! emptyString(parameters)) {
     ndbout << "No parameters expected to this command." << endl;
-    return;
+    return -1;
   }
 
 #if 0
@@ -1893,19 +1952,21 @@ void CommandInterpreter::executeGetStat(int /*processId*/,
   ndbout << "Number of GETSTAT commands: " 
   << statistics._test1 << endl;
   */
+  return 0;
 }
 
 //*****************************************************************************
 //*****************************************************************************
 				 
-void 
+int
 CommandInterpreter::executeEventReporting(int processId,
 					  const char* parameters, 
 					  bool all) 
 {
+  int retval = 0;
   if (emptyString(parameters)) {
     ndbout << "Expected argument" << endl;
-    return;
+    return -1;
   }
   BaseString tmp(parameters);
   Vector<BaseString> specs;
@@ -1952,10 +2013,12 @@ CommandInterpreter::executeEventReporting(int processId,
     if (result != 0) {
       ndbout_c(" failed."); 
       printError();
+      retval = -1;
     } else {
       ndbout_c(" OK!"); 
     }
   }
+  return retval;
 }
 
 /*****************************************************************************
@@ -2056,7 +2119,7 @@ CommandInterpreter::executeStartBackup(char* parameters)
   return 0;
 }
 
-void
+int
 CommandInterpreter::executeAbortBackup(char* parameters) 
 {
   int bid = -1;
@@ -2075,14 +2138,15 @@ CommandInterpreter::executeAbortBackup(char* parameters)
     if (result != 0) {
       ndbout << "Abort of backup " << bid << " failed" << endl;
       printError();
+      return -1;
     } else {
       ndbout << "Abort of backup " << bid << " ordered" << endl;
     }
   }
-  return;
+  return 0;
  executeAbortBackupError1:
   ndbout << "Invalid arguments: expected <BackupId>" << endl;
-  return;
+  return -1;
 }
 
 #ifdef HAVE_GLOBAL_REPLICATION
@@ -2113,12 +2177,12 @@ CommandInterpreter::executeAbortBackup(char* parameters)
 
  *****************************************************************************/
 
-void
+int
 CommandInterpreter::executeRep(char* parameters) 
 {
   if (emptyString(parameters)) {
     ndbout << helpTextRep;
-    return;
+    return 0;
   }
 
   char * line = my_strdup(parameters,MYF(MY_WME));
@@ -2138,7 +2202,7 @@ CommandInterpreter::executeRep(char* parameters)
     if(host == NULL)
     {
       ndbout_c("host:port must be specified.");
-      return;
+      return -1;
     }
     
     if(rep_connected) {
@@ -2150,14 +2214,17 @@ CommandInterpreter::executeRep(char* parameters)
           
     if(m_repserver == NULL)
       m_repserver = ndb_rep_create_handle();
-    if(ndb_rep_connect(m_repserver, host) < 0)
-      ndbout_c("Failed to connect to %s", host); 
+    if(ndb_rep_connect(m_repserver, host) < 0){
+      ndbout_c("Failed to connect to %s", host);
+      return -1;
+    } 
     else
       rep_connected=true;
-    return;
+    return 0;
     
     if(!rep_connected) {
       ndbout_c("Not connected to REP server");
+      return -1;
     }
   }
     
@@ -2191,17 +2258,18 @@ CommandInterpreter::executeRep(char* parameters)
       req = GrepReq::START_DELETE;
     } else {
       ndbout_c("Illegal argument to command 'REPLICATION START'");
-      return;
+      return -1;
     }
 
     int result = ndb_rep_command(m_repserver, req, &repId, &reply);
     
     if (result != 0) {
       ndbout << "Start of Global Replication failed" << endl;
+      return -1;
     } else {
       ndbout << "Start of Global Replication ordered" << endl;
     }
-    return;
+    return 0;
   }
 
   /********
@@ -2221,7 +2289,7 @@ CommandInterpreter::executeRep(char* parameters)
       char *strEpoch = strtok(NULL, "\0");
       if(strEpoch == NULL) {
 	ndbout_c("Epoch expected!");
-	return;
+	return -1;
       }
       req = GrepReq::STOP;
       epoch=atoi(strEpoch);      
@@ -2245,16 +2313,17 @@ CommandInterpreter::executeRep(char* parameters)
       req = GrepReq::STOP_DELETE;
     } else {
       ndbout_c("Illegal argument to command 'REPLICATION STOP'");
-      return;
+      return -1;
     }
     int result = ndb_rep_command(m_repserver, req, &repId, &reply, epoch);
     
     if (result != 0) {
       ndbout << "Stop command failed" << endl;
+      return -1;
     } else {
       ndbout << "Stop ordered" << endl;
     }
-    return;
+    return 0;
   }
 
   /*********
@@ -2267,6 +2336,7 @@ CommandInterpreter::executeRep(char* parameters)
     
     if (result != 0) {
       ndbout << "Status request of Global Replication failed" << endl;
+      return -1;
     } else {
       ndbout << "Status request of Global Replication ordered" << endl;
       ndbout << "See printout at one of the DB nodes" << endl;
@@ -2274,7 +2344,7 @@ CommandInterpreter::executeRep(char* parameters)
       ndbout << " SubscriptionId " << repstate.subid 
 	     << " SubscriptionKey " << repstate.subkey << endl;
     }
-    return;
+    return 0;
   }
 
   /*********
@@ -2293,6 +2363,7 @@ CommandInterpreter::executeRep(char* parameters)
     
     if (result != 0) {
       ndbout << "Query repserver failed" << endl;
+      return -1;
     } else {
       ndbout << "Query repserver sucessful" << endl;
       ndbout_c("repstate : QueryCounter %d, f=%d l=%d"
@@ -2301,8 +2372,9 @@ CommandInterpreter::executeRep(char* parameters)
 	       repstate.first[0], repstate.last[0],
 	       repstate.no_of_nodegroups );
     }
-    return;
+    return 0;
   }
+  return 0;
 }
 #endif // HAVE_GLOBAL_REPLICATION
 
diff --git a/ndb/test/ndbapi/flexScan.cpp b/ndb/test/ndbapi/flexScan.cpp
index 4d2c85d6955..1f001bd0210 100644
--- a/ndb/test/ndbapi/flexScan.cpp
+++ b/ndb/test/ndbapi/flexScan.cpp
@@ -27,7 +27,7 @@
 	 verify delete
 
        Arguments:
-        -f Location of Ndb.cfg file, default Ndb.cfg
+        -f Location of my.cnf file, default my.cnf
         -t Number of threads to start, default 1
         -o Number of operations per loop, default 500	-l Number of loops to run, default 1, 0=infinite
         -a Number of attributes, default 25
@@ -829,7 +829,7 @@ static int createTables(Ndb* pMyNdb)
 static void printUsage()
 {
   ndbout << "Usage of flexScan:" << endl;
-  ndbout << "-f <path> Location of Ndb.cfg file, default: Ndb.cfg" << endl;
+  ndbout << "-f <path> Location of my.cnf file, default: my.cnf" << endl;
   ndbout << "-t <int>  Number of threads to start, default 1" << endl;
   ndbout << "-o <int>  Number of operations per loop, default 500" << endl;
   ndbout << "-l <int>  Number of loops to run, default 1, 0=infinite" << endl;
diff --git a/ndb/tools/ndb_config.cpp b/ndb/tools/ndb_config.cpp
index 78a2fa38fba..0eaabf76eae 100644
--- a/ndb/tools/ndb_config.cpp
+++ b/ndb/tools/ndb_config.cpp
@@ -64,7 +64,7 @@ static struct my_option my_long_options[] =
   { "ndb-connectstring", 256,
     "Set connect string for connecting to ndb_mgmd. " 
     "Syntax: \"[nodeid=<id>;][host=]<hostname>[:<port>]\". " 
-    "Overides specifying entries in NDB_CONNECTSTRING and Ndb.cfg", 
+    "Overrides specifying entries in NDB_CONNECTSTRING and my.cnf", 
     (gptr*) &g_connectstring, (gptr*) &g_connectstring, 
     0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
   { "nodes", 256, "Print nodes",
diff --git a/ndb/tools/ndb_size.pl b/ndb/tools/ndb_size.pl
index c285a7590fd..3d1ea3f4231 100644
--- a/ndb/tools/ndb_size.pl
+++ b/ndb/tools/ndb_size.pl
@@ -57,7 +57,7 @@ if(@ARGV < 3 || $ARGV[0] eq '--usage' || $ARGV[0] eq '--help')
     $template->param(dsn => $dsn);
 }
 
-my @releases = ({rel=>'4.1'},{rel=>'5.0'},{rel=>'5.1'});
+my @releases = ({rel=>'4.1'},{rel=>'5.0'},{rel=>'5.1'}); #,{rel=>'5.1-dd'});
 $template->param(releases => \@releases);
 
 my $tables  = $dbh->selectall_arrayref("show tables");
@@ -81,25 +81,29 @@ sub align {
     return @aligned;
 }
 
-foreach(@{$tables})
-{
-    my $table= @{$_}[0];
+sub do_table {
+    my $table= shift;
+    my $info= shift;
+    my %indexes= %{$_[0]};
+    my @count= @{$_[1]};
+
     my @columns;
-    my $info= $dbh->selectall_hashref('describe `'.$table.'`',"Field");
-    my @count  = $dbh->selectrow_array('select count(*) from `'.$table.'`');
     my %columnsize; # used for index calculations
-
     # We now work out the DataMemory usage
     
-    # sizes for   4.1, 5.0, 5.1
-    my @totalsize= (0,0,0);
+    # sizes for   4.1, 5.0, 5.1 and 5.1-dd
+    my @totalsize= (0,0,0,0);
+    @totalsize= @totalsize[0..$#releases]; # limit to releases we're outputting
+    my $nrvarsize= 0;
 
     foreach(keys %$info)
     {
-	my @realsize = (0,0,0);
+	my @realsize = (0,0,0,0);
+	my @varsize  = (0,0,0,0);
 	my $type;
 	my $size;
 	my $name= $_;
+	my $is_varsize= 0;
 
 	if($$info{$_}{Type} =~ /^(.*?)\((\d+)\)/)
 	{
@@ -112,54 +116,86 @@ foreach(@{$tables})
 	}
 
 	if($type =~ /tinyint/)
-	{@realsize=(1,1,1)}
+	{@realsize=(1,1,1,1)}
 	elsif($type =~ /smallint/)
-	{@realsize=(2,2,2)}
+	{@realsize=(2,2,2,2)}
 	elsif($type =~ /mediumint/)
-	{@realsize=(3,3,3)}
+	{@realsize=(3,3,3,3)}
 	elsif($type =~ /bigint/)
-	{@realsize=(8,8,8)}
+	{@realsize=(8,8,8,8)}
 	elsif($type =~ /int/)
-	{@realsize=(4,4,4)}
+	{@realsize=(4,4,4,4)}
 	elsif($type =~ /float/)
 	{
 	    if($size<=24)
-	    {@realsize=(4,4,4)}
+	    {@realsize=(4,4,4,4)}
 	    else
-	    {@realsize=(8,8,8)}
+	    {@realsize=(8,8,8,8)}
 	}
 	elsif($type =~ /double/ || $type =~ /real/)
-	{@realsize=(8,8,8)}
+	{@realsize=(8,8,8,8)}
 	elsif($type =~ /bit/)
 	{
 	    my $a=($size+7)/8;
-	    @realsize = ($a,$a,$a);
+	    @realsize = ($a,$a,$a,$a);
 	}
 	elsif($type =~ /datetime/)
-	{@realsize=(8,8,8)}
+	{@realsize=(8,8,8,8)}
 	elsif($type =~ /timestamp/)
-	{@realsize=(4,4,4)}
+	{@realsize=(4,4,4,4)}
 	elsif($type =~ /date/ || $type =~ /time/)
-	{@realsize=(3,3,3)}
+	{@realsize=(3,3,3,3)}
 	elsif($type =~ /year/)
-	{@realsize=(1,1,1)}
+	{@realsize=(1,1,1,1)}
 	elsif($type =~ /varchar/ || $type =~ /varbinary/)
 	{
-	    my $fixed= 1+$size;
+	    my $fixed=$size+ceil($size/256);
 	    my @dynamic=$dbh->selectrow_array("select avg(length(`"
 					      .$name
 					      ."`)) from `".$table.'`');
 	    $dynamic[0]=0 if !$dynamic[0];
-	    @realsize= ($fixed,$fixed,ceil($dynamic[0]));
+	    $dynamic[0]+=ceil($dynamic[0]/256); # size bit
+	    $nrvarsize++;
+	    $is_varsize= 1;
+	    $varsize[3]= ceil($dynamic[0]);
+	    @realsize= ($fixed,$fixed,ceil($dynamic[0]),$fixed);
 	}
 	elsif($type =~ /binary/ || $type =~ /char/)
-	{@realsize=($size,$size,$size)}
+	{@realsize=($size,$size,$size,$size)}
 	elsif($type =~ /text/ || $type =~ /blob/)
 	{
-	    @realsize=(256,256,1);
-	    $NoOfTables[$_]{val} += 1 foreach 0..$#releases; # blob uses table
-	} # FIXME check if 5.1 is correct
+	    @realsize=(8+256,8+256,8+256,8+256);
+
+	    my $blobhunk= 2000;
+	    $blobhunk= 8000 if $type=~ /longblob/;
+	    $blobhunk= 4000 if $type=~ /mediumblob/;
+
+	    my @blobsize=$dbh->selectrow_array("select SUM(CEILING(".
+					       "length(`$name`)/$blobhunk))".
+					       "from `".$table."`");
+	    $blobsize[0]=0 if !defined($blobsize[0]);
+	    #$NoOfTables[$_]{val} += 1 foreach 0..$#releases; # blob uses table
+	    do_table($table."\$BLOB_$name",
+		     {'PK'=>{Type=>'int'},
+		      'DIST'=>{Type=>'int'},
+		      'PART'=>{Type=>'int'},
+		      'DATA'=>{Type=>"binary($blobhunk)"}
+		  },
+		     {'PRIMARY' => {
+                         'unique' => 1,
+                         'comment' => '',
+                         'columns' => [
+                                        'PK',
+				        'DIST',
+				        'PART',
+                                      ],
+			 'type' => 'HASH'
+				    }
+		  },
+		     \@blobsize);
+	}
 
+	@realsize= @realsize[0..$#releases];
 	@realsize= align(4,@realsize);
 
 	$totalsize[$_]+=$realsize[$_] foreach 0..$#totalsize;
@@ -170,6 +206,7 @@ foreach(@{$tables})
 	push @columns, {
 	    name=>$name,
 	    type=>$type,
+	    is_varsize=>$is_varsize,
 	    size=>$size,
 	    key=>$$info{$_}{Key},
 	    datamemory=>\@realout,
@@ -183,24 +220,10 @@ foreach(@{$tables})
     # Firstly, we assemble some information about the indexes.
     # We use SHOW INDEX instead of using INFORMATION_SCHEMA so
     # we can still connect to pre-5.0 mysqlds.
-    my %indexes;
-    {
-	my $sth= $dbh->prepare("show index from `".$table.'`');
-	$sth->execute;
-	while(my $i = $sth->fetchrow_hashref)
-	{    
-	    $indexes{${%$i}{Key_name}}= {
-		type=>${%$i}{Index_type},
-		unique=>!${%$i}{Non_unique},
-		comment=>${%$i}{Comment},
-	    } if !defined($indexes{${%$i}{Key_name}});
-
-	    $indexes{${%$i}{Key_name}}{columns}[${%$i}{Seq_in_index}-1]= 
-		${%$i}{Column_name};
-	}
-    }
 
     if(!defined($indexes{PRIMARY})) {
+	my @usage= ({val=>8},{val=>8},{val=>8},{val=>8});
+	@usage= @usage[0..$#releases];
 	$indexes{PRIMARY}= {
 	    type=>'BTREE',
 	    unique=>1,
@@ -212,20 +235,22 @@ foreach(@{$tables})
 	    type=>'bigint',
 	    size=>8,
 	    key=>'PRI',
-	    datamemory=>[{val=>8},{val=>8},{val=>8}],
+	    datamemory=>\@usage,
 	};
 	$columnsize{'HIDDEN_NDB_PKEY'}= [8,8,8];
     }
 
-    my @IndexDataMemory= ({val=>0},{val=>0},{val=>0});
-    my @RowIndexMemory= ({val=>0},{val=>0},{val=>0});
+    my @IndexDataMemory= ({val=>0},{val=>0},{val=>0},{val=>0});
+    my @RowIndexMemory= ({val=>0},{val=>0},{val=>0},{val=>0});
+    @IndexDataMemory= @IndexDataMemory[0..$#releases];
+    @RowIndexMemory= @RowIndexMemory[0..$#releases];
 
     my @indexes;
     foreach my $index (keys %indexes) {
 	my $im41= 25;
 	$im41+=$columnsize{$_}[0] foreach @{$indexes{$index}{columns}};
-	my @im = ({val=>$im41},{val=>25},{val=>25});
-	my @dm = ({val=>10},{val=>10},{val=>10});
+	my @im = ({val=>$im41},{val=>25},{val=>25}); #,{val=>25});
+	my @dm = ({val=>10},{val=>10},{val=>10}); #,{val=>10});
 	push @indexes, {
 	    name=>$index,
 	    type=>$indexes{$index}{type},
@@ -233,13 +258,22 @@ foreach(@{$tables})
 	    indexmemory=>\@im,
 	    datamemory=>\@dm,
 	};
-	$IndexDataMemory[$_]{val}+=$dm[$_]{val} foreach 0..2;
-	$RowIndexMemory[$_]{val}+=$im[$_]{val} foreach 0..2;
+	$IndexDataMemory[$_]{val}+=$dm[$_]{val} foreach 0..$#releases;
+	$RowIndexMemory[$_]{val}+=$im[$_]{val} foreach 0..$#releases;
     }
 
     # total size + 16 bytes overhead
     my @TotalDataMemory;
-    $TotalDataMemory[$_]{val}=$IndexDataMemory[$_]{val}+$totalsize[$_]+16 foreach 0..2;
+    my @RowOverhead = ({val=>16},{val=>16},{val=>16}); #,{val=>24});
+    # 5.1 has ptr to varsize page, and per-varsize overhead
+    my @nrvarsize_mem= ({val=>0},{val=>0},
+			{val=>8}); #,{val=>0});
+    {
+	my @a= align(4,$nrvarsize*2);
+	$nrvarsize_mem[2]{val}+=$a[0]+$nrvarsize*4;
+    }
+
+    $TotalDataMemory[$_]{val}=$IndexDataMemory[$_]{val}+$totalsize[$_]+$RowOverhead[$_]{val}+$nrvarsize_mem[$_]{val} foreach 0..$#releases;
 
     my @RowDataMemory;
     push @RowDataMemory,{val=>$_} foreach @totalsize;
@@ -260,12 +294,18 @@ foreach(@{$tables})
     my @counts;
     $counts[$_]{val}= $count foreach 0..$#releases;
 
+    my @nrvarsize_rel= ({val=>0},{val=>0},
+			{val=>$nrvarsize}); #,{val=>0});
+
     push @table_size, {
 	table=>$table,
 	indexes=>\@indexes,
 	columns=>\@columns,
 	count=>\@counts,
+	RowOverhead=>\@RowOverhead,
 	RowDataMemory=>\@RowDataMemory,
+	nrvarsize=>\@nrvarsize_rel,
+	nrvarsize_mem=>\@nrvarsize_mem,
 	releases=>\@releases,
 	IndexDataMemory=>\@IndexDataMemory,
 	TotalDataMemory=>\@TotalDataMemory,
@@ -283,6 +323,31 @@ foreach(@{$tables})
     $NoOfIndexes[$_]{val} += @indexes foreach 0..$#releases;
 }
 
+foreach(@{$tables})
+{
+    my $table= @{$_}[0];
+    my $info= $dbh->selectall_hashref('describe `'.$table.'`',"Field");
+    my @count  = $dbh->selectrow_array('select count(*) from `'.$table.'`');
+
+    my %indexes;
+    {
+	my $sth= $dbh->prepare("show index from `".$table.'`');
+	$sth->execute;
+	while(my $i = $sth->fetchrow_hashref)
+	{    
+	    $indexes{${%$i}{Key_name}}= {
+		type=>${%$i}{Index_type},
+		unique=>!${%$i}{Non_unique},
+		comment=>${%$i}{Comment},
+	    } if !defined($indexes{${%$i}{Key_name}});
+
+	    $indexes{${%$i}{Key_name}}{columns}[${%$i}{Seq_in_index}-1]= 
+		${%$i}{Column_name};
+	}
+    }
+    do_table($table, $info, \%indexes, \@count);
+}
+
 my @NoOfTriggers;
 # for unique hash indexes
 $NoOfTriggers[$_]{val} += $NoOfIndexes[$_]{val}*3 foreach 0..$#releases;
diff --git a/ndb/tools/ndb_size.tmpl b/ndb/tools/ndb_size.tmpl
index dc02b5a5970..1e19ea132ba 100644
--- a/ndb/tools/ndb_size.tmpl
+++ b/ndb/tools/ndb_size.tmpl
@@ -15,6 +15,8 @@ td,th { border: 1px solid black }
 
 <p>This information should be valid for MySQL 4.1 and 5.0. Since 5.1 is not a final release yet, the numbers should be used as a guide only.</p>
 
+<p>5.1-dd is for tables stored on disk. The ndb_size.pl estimates are <b>experimental</b> and should not be trusted. Notably we don't take into account indexed columns being in DataMemory versus non-indexed on disk.</p>
+
 <h2>Parameter Settings</h2>
 <p><b>NOTE</b> the configuration parameters below do not take into account system tables and other requirements.</p>
 <table>
@@ -69,6 +71,7 @@ td,th { border: 1px solid black }
   <tr>
     <th>Column</th>
     <th>Type</th>
+    <th>VARSIZE</th>
     <th>Size</th>
     <th>Key</th>
     <TMPL_LOOP NAME=releases>
@@ -79,6 +82,7 @@ td,th { border: 1px solid black }
     <tr>
       <td><TMPL_VAR NAME=name></td>
       <td><TMPL_VAR NAME=type></td>
+      <td><TMPL_IF NAME=is_varsize>YES<TMPL_ELSE>&nbsp;</TMPL_IF></td>
       <td><TMPL_VAR NAME=size></td>
       <td><TMPL_VAR NAME=key></td>
       <TMPL_LOOP NAME=datamemory>
@@ -129,9 +133,21 @@ td,th { border: 1px solid black }
  </TMPL_LOOP>
 </tr>
 <tr>
+ <th>Nr Varsized Attributes</th>
+ <TMPL_LOOP NAME=nrvarsize>
+  <td><TMPL_VAR NAME=val></td>
+ </TMPL_LOOP>
+</tr>
+<tr>
  <th>Row Overhead</th>
- <TMPL_LOOP NAME=releases>
-  <td>16</td>
+ <TMPL_LOOP NAME=RowOverhead>
+  <td><TMPL_VAR NAME=val></td>
+ </TMPL_LOOP>
+</tr>
+<tr>
+ <th>Varsized Overhead</th>
+ <TMPL_LOOP NAME=nrvarsize_mem>
+  <td><TMPL_VAR NAME=val></td>
  </TMPL_LOOP>
 </tr>
 <tr>
diff --git a/ndb/tools/restore/consumer.hpp b/ndb/tools/restore/consumer.hpp
index 692c814159f..6a8ef29e295 100644
--- a/ndb/tools/restore/consumer.hpp
+++ b/ndb/tools/restore/consumer.hpp
@@ -31,6 +31,7 @@ public:
   virtual void logEntry(const LogEntry &){}
   virtual void endOfLogEntrys(){}
   virtual bool finalize_table(const TableS &){return true;}
+  virtual bool has_temp_error() {return false;}
 };
 
 #endif
diff --git a/ndb/tools/restore/consumer_restore.cpp b/ndb/tools/restore/consumer_restore.cpp
index 70ea7460d78..14c02b66bfc 100644
--- a/ndb/tools/restore/consumer_restore.cpp
+++ b/ndb/tools/restore/consumer_restore.cpp
@@ -140,6 +140,11 @@ BackupRestore::finalize_table(const TableS & table){
 }
 
 bool
+BackupRestore::has_temp_error(){
+  return m_temp_error;
+}
+
+bool
 BackupRestore::table(const TableS & table){
   if (!m_restore && !m_restore_meta)
     return true;
@@ -437,6 +442,7 @@ bool BackupRestore::errorHandler(restore_callback_t *cb)
     
   case NdbError::TemporaryError:
     err << "Temporary error: " << error << endl;
+    m_temp_error = true;
     NdbSleep_MilliSleep(sleepTime);
     return true;
     // RETRY
diff --git a/ndb/tools/restore/consumer_restore.hpp b/ndb/tools/restore/consumer_restore.hpp
index df219cd4412..261b108b184 100644
--- a/ndb/tools/restore/consumer_restore.hpp
+++ b/ndb/tools/restore/consumer_restore.hpp
@@ -41,6 +41,7 @@ public:
     m_parallelism = parallelism;
     m_callback = 0;
     m_free_callback = 0;
+    m_temp_error = false;
     m_transactions = 0;
     m_cache.m_old_table = 0;
   }
@@ -60,6 +61,7 @@ public:
   virtual void logEntry(const LogEntry &);
   virtual void endOfLogEntrys();
   virtual bool finalize_table(const TableS &);
+  virtual bool has_temp_error();
   void connectToMysql();
   Ndb * m_ndb;
   bool m_restore;
@@ -72,6 +74,7 @@ public:
 
   restore_callback_t *m_callback;
   restore_callback_t *m_free_callback;
+  bool m_temp_error;
 
   /**
    * m_new_table_ids[X] = Y;
diff --git a/ndb/tools/restore/restore_main.cpp b/ndb/tools/restore/restore_main.cpp
index d786dffe89e..779447306af 100644
--- a/ndb/tools/restore/restore_main.cpp
+++ b/ndb/tools/restore/restore_main.cpp
@@ -411,6 +411,17 @@ main(int argc, char** argv)
       }
     }
   }
+  for(Uint32 i= 0; i < g_consumers.size(); i++) 
+  {
+    if (g_consumers[i]->has_temp_error())
+    {
+      clearConsumers();
+      ndbout_c("\nRestore successful, but encountered temporary error, "
+               "please look at configuration.");
+      return NDBT_ProgramExit(NDBT_TEMPORARY);
+    }
+  }
+
   clearConsumers();
   return NDBT_ProgramExit(NDBT_OK);
 } // main