Merge mleich@bk-internal.mysql.com:/home/bk/mysql-4.1

into three.local.lan:/home/matthias/Arbeit/mysql-4.1/src

configure.in

@@ -16,7 +16,7 @@ SHARED_LIB_VERSION=14:0:0
 # ndb version
 NDB_VERSION_MAJOR=4
 NDB_VERSION_MINOR=1
-NDB_VERSION_BUILD=8
+NDB_VERSION_BUILD=9
 NDB_VERSION_STATUS=""
 
 # Set all version vars based on $VERSION. How do we do this more elegant ?

mysql-test/r/rpl_insert_ignore.result

+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+CREATE TABLE t1 (
+a int unsigned not null auto_increment primary key,
+b int unsigned,
+unique (b)
+) ENGINE=innodb;
+CREATE TABLE t2 (
+a int unsigned, # to force INSERT SELECT to have a certain order
+b int unsigned
+) ENGINE=innodb;
+INSERT INTO t1 VALUES (NULL, 1);
+INSERT INTO t1 VALUES (NULL, 2);
+INSERT INTO t1 VALUES (NULL, 3);
+INSERT INTO t1 VALUES (NULL, 4);
+INSERT INTO t2 VALUES (1, 1);
+INSERT INTO t2 VALUES (2, 2);
+INSERT INTO t2 VALUES (3, 5);
+INSERT INTO t2 VALUES (4, 3);
+INSERT INTO t2 VALUES (5, 4);
+INSERT INTO t2 VALUES (6, 6);
+INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	1
+2	2
+3	3
+4	4
+5	5
+6	6
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	1
+2	2
+3	3
+4	4
+5	5
+6	6
+drop table t1;
+CREATE TABLE t1 (
+a int unsigned not null auto_increment primary key,
+b int unsigned,
+unique (b)
+) ENGINE=myisam;
+INSERT INTO t1 VALUES (1, 1);
+INSERT INTO t1 VALUES (2, 2);
+INSERT INTO t1 VALUES (3, 3);
+INSERT INTO t1 VALUES (4, 4);
+INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	1
+2	2
+3	3
+4	4
+5	5
+6	6
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	1
+2	2
+3	3
+4	4
+5	5
+6	6
+drop table t1, t2;

mysql-test/t/rpl_insert_ignore-slave.opt

+--innodb

mysql-test/t/rpl_insert_ignore.test

+# Testcase for BUG#6287 "Slave skips auto_increment values in Replication with InnoDB"
+# The bug was that if on master, INSERT IGNORE ignored some
+# rows, and the table was InnoDB with auto_inc column, then on slave
+# some rows received an auto_inc bigger than on master.
+# Slave needs to be started with --innodb to store table in InnoDB.
+# Same test for MyISAM (which had no bug).
+
+-- source include/have_innodb.inc
+
+-- source include/master-slave.inc
+
+CREATE TABLE t1 (
+ a int unsigned not null auto_increment primary key,
+ b int unsigned,
+ unique (b)
+) ENGINE=innodb;
+
+CREATE TABLE t2 (
+ a int unsigned, # to force INSERT SELECT to have a certain order
+ b int unsigned
+) ENGINE=innodb;
+
+
+INSERT INTO t1 VALUES (NULL, 1);
+INSERT INTO t1 VALUES (NULL, 2);
+INSERT INTO t1 VALUES (NULL, 3);
+INSERT INTO t1 VALUES (NULL, 4);
+
+# An alternation of values which will conflict in t1 and will not.
+
+INSERT INTO t2 VALUES (1, 1);
+INSERT INTO t2 VALUES (2, 2);
+INSERT INTO t2 VALUES (3, 5);
+INSERT INTO t2 VALUES (4, 3);
+INSERT INTO t2 VALUES (5, 4);
+INSERT INTO t2 VALUES (6, 6);
+
+INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
+
+# Compare results
+
+SELECT * FROM t1 ORDER BY a;
+
+sync_slave_with_master;
+SELECT * FROM t1 ORDER BY a;
+
+# Now do the same for MyISAM
+
+connection master;
+drop table t1;
+CREATE TABLE t1 (
+ a int unsigned not null auto_increment primary key,
+ b int unsigned,
+ unique (b)
+) ENGINE=myisam;
+
+INSERT INTO t1 VALUES (1, 1);
+INSERT INTO t1 VALUES (2, 2);
+INSERT INTO t1 VALUES (3, 3);
+INSERT INTO t1 VALUES (4, 4);
+
+INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
+
+SELECT * FROM t1 ORDER BY a;
+
+sync_slave_with_master;
+SELECT * FROM t1 ORDER BY a;
+
+connection master;
+drop table t1, t2;
+sync_slave_with_master;

ndb/include/ndbapi/Ndb.hpp

@@ -1586,7 +1586,6 @@ private:
 /******************************************************************************
  *	These are the private variables in this class.	
  *****************************************************************************/
-  NdbObjectIdMap*       theNdbObjectIdMap;
   Ndb_cluster_connection   *m_ndb_cluster_connection;
 
   NdbConnection**       thePreparedTransactionsArray;
@@ -1637,10 +1636,6 @@ private:
   Uint32   theMyRef;        // My block reference  
   Uint32   theNode;         // The node number of our node
   
-  Uint32   theNoOfDBnodes;  // The number of DB nodes  
-  Uint32 * theDBnodes;      // The node number of the DB nodes
-  Uint8    *the_release_ind;// 1 indicates to release all connections to node 
-  
   Uint64               the_last_check_time;
   Uint64               theFirstTransId;
   
@@ -1663,10 +1658,6 @@ private:
     InitConfigError
   } theInitState;
 
-  // Ensure good distribution of connects
-  Uint32		theCurrentConnectIndex;
-  Uint32		theCurrentConnectCounter;
-  
   /**
    * Computes fragement id for primary key
    *
@@ -1692,7 +1683,7 @@ private:
     Uint32 noOfFragments;
     Uint32 * fragment2PrimaryNodeMap;
     
-    void init(Uint32 noOfNodes, Uint32 nodeIds[]);
+    void init(Uint32 noOfNodes, Uint8 nodeIds[]);
     void release();
   } startTransactionNodeSelectionData;
   

ndb/src/common/util/version.c

@@ -17,6 +17,8 @@
 #include <ndb_global.h>
 #include <ndb_version.h>
 #include <version.h>
+#include <NdbEnv.h>
+#include <NdbOut.hpp>
 
 Uint32 getMajor(Uint32 version) {
   return (version >> 16) & 0xFF;
@@ -67,8 +69,27 @@ struct NdbUpGradeCompatible {
 
 /*#define TEST_VERSION*/
 
+#define HAVE_NDB_SETVERSION
+#ifdef HAVE_NDB_SETVERSION
+Uint32 ndbOwnVersionTesting = 0;
+void
+ndbSetOwnVersion() {
+  char buf[256];
+  if (NdbEnv_GetEnv("NDB_SETVERSION", buf, sizeof(buf))) {
+    Uint32 _v1,_v2,_v3;
+    if (sscanf(buf, "%u.%u.%u", &_v1, &_v2, &_v3) == 3) {
+      ndbOwnVersionTesting = MAKE_VERSION(_v1,_v2,_v3);
+      ndbout_c("Testing: Version set to 0x%x",  ndbOwnVersionTesting);
+    }
+  }
+}
+#else
+void ndbSetOwnVersion() {}
+#endif
+
 #ifndef TEST_VERSION
 struct NdbUpGradeCompatible ndbCompatibleTable_full[] = {
+  { MAKE_VERSION(4,1,9), MAKE_VERSION(4,1,8), UG_Exact },
   { MAKE_VERSION(3,5,2), MAKE_VERSION(3,5,1), UG_Exact },
   { 0, 0, UG_Null }
 };
@@ -78,8 +99,6 @@ struct NdbUpGradeCompatible ndbCompatibleTable_upgrade[] = {
   { 0, 0, UG_Null }
 };
 
-void ndbSetOwnVersion() {}
-
 #else /* testing purposes */
 
 struct NdbUpGradeCompatible ndbCompatibleTable_full[] = {
@@ -100,19 +119,6 @@ struct NdbUpGradeCompatible ndbCompatibleTable_upgrade[] = {
 };
 
 
-Uint32 ndbOwnVersionTesting = 0;
-void
-ndbSetOwnVersion() {
-  char buf[256];
-  if (NdbEnv_GetEnv("NDB_SETVERSION", buf, sizeof(buf))) {
-    Uint32 _v1,_v2,_v3;
-    if (sscanf(buf, "%u.%u.%u", &_v1, &_v2, &_v3) == 3) {
-      ndbOwnVersionTesting = MAKE_VERSION(_v1,_v2,_v3);
-      ndbout_c("Testing: Version set to 0x%x",  ndbOwnVersionTesting);
-    }
-  }
-}
-
 #endif
 
 void ndbPrintVersion()
@@ -126,13 +132,13 @@ void ndbPrintVersion()
 Uint32
 ndbGetOwnVersion()
 {
-#ifndef TEST_VERSION
-  return NDB_VERSION_D;
-#else /* testing purposes */
+#ifdef HAVE_NDB_SETVERSION
   if (ndbOwnVersionTesting == 0)
     return NDB_VERSION_D;
   else
     return ndbOwnVersionTesting;
+#else
+  return NDB_VERSION_D;
 #endif
 }
 

ndb/src/mgmsrv/CommandInterpreter.hpp

@@ -130,7 +130,6 @@ public:
   void executeStop(int processId, const char* parameters, bool all);
   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 executeTrace(int processId, const char* parameters, bool all);
   void executeLog(int processId, const char* parameters, bool all);
@@ -140,14 +139,7 @@ public:
   void executeTestOn(int processId, const char* parameters, bool all);
   void executeTestOff(int processId, const char* parameters, bool all);
   void executeStatus(int processId, const char* parameters, bool all);
-  void executeEnterSingleUser(char* parameters);
-  void executeExitSingleUser(char* parameters);
-  void executeEventReporting(int processId, const char* parameters, bool all);
   void executeDumpState(int processId, const char* parameters, bool all);
-  void executeStartBackup(char * pars);
-  void executeAbortBackup(char * pars);
-
-  void jonas(int processId, const char* parameters, bool all);  
 
   /**
    * A execute function definition

ndb/src/ndbapi/Ndb.cpp

@@ -65,37 +65,25 @@ NdbConnection* Ndb::doConnect(Uint32 tConNode)
 // We will connect to any node. Make sure that we have connections to all
 // nodes.
 //****************************************************************************
-  Uint32 tNoOfDbNodes = theNoOfDBnodes;
-  i = theCurrentConnectIndex;
+  Uint32 tNoOfDbNodes= theImpl->theNoOfDBnodes;
+  Uint32 &theCurrentConnectIndex= theImpl->theCurrentConnectIndex;
   UintR Tcount = 0;
   do {
-    if (i >= tNoOfDbNodes) {
-      i = 0;
+    theCurrentConnectIndex++;
+    if (theCurrentConnectIndex >= tNoOfDbNodes) {
+      theCurrentConnectIndex = 0;
     }//if
     Tcount++;
-    tNode = theDBnodes[i];
+    tNode = theImpl->theDBnodes[theCurrentConnectIndex];
     TretCode = NDB_connect(tNode);
     if ((TretCode == 1) || (TretCode == 2)) {
 //****************************************************************************
 // We have connections now to the desired node. Return
 //****************************************************************************
-      if (theCurrentConnectIndex == i) {
-        theCurrentConnectCounter++;
-        if (theCurrentConnectCounter == 8) {
-	  theCurrentConnectCounter = 1;
-	  theCurrentConnectIndex++;
-	}//if
-      } else {
-	// Set to 2 because we have already connected to a node 
-	// when we get here.
-        theCurrentConnectCounter = 2;
-        theCurrentConnectIndex = i;
-      }//if
       return getConnectedNdbConnection(tNode);
     } else if (TretCode != 0) {
       tAnyAlive = 1;
     }//if
-    i++;
   } while (Tcount < tNoOfDbNodes);
 //****************************************************************************
 // We were unable to find a free connection. If no node alive we will report
@@ -211,8 +199,9 @@ Ndb::doDisconnect()
   NdbConnection* tNdbCon;
   CHECK_STATUS_MACRO_VOID;
 
-  DBUG_PRINT("info", ("theNoOfDBnodes=%d", theNoOfDBnodes));
-  Uint32 tNoOfDbNodes = theNoOfDBnodes;
+  Uint32 tNoOfDbNodes = theImpl->theNoOfDBnodes;
+  Uint8 *theDBnodes= theImpl->theDBnodes;
+  DBUG_PRINT("info", ("theNoOfDBnodes=%d", tNoOfDbNodes));
   UintR i;
   for (i = 0; i < tNoOfDbNodes; i++) {
     Uint32 tNode = theDBnodes[i];
@@ -259,8 +248,8 @@ Ndb::waitUntilReady(int timeout)
       unsigned int foundAliveNode = 0;
       TransporterFacade *tp = TransporterFacade::instance();
       tp->lock_mutex();
-      for (unsigned int i = 0; i < theNoOfDBnodes; i++) {
-	const NodeId nodeId = theDBnodes[i];
+      for (unsigned int i = 0; i < theImpl->theNoOfDBnodes; i++) {
+	const NodeId nodeId = theImpl->theDBnodes[i];
 	//************************************************
 	// If any node is answering, ndb is answering
 	//************************************************
@@ -270,7 +259,7 @@ Ndb::waitUntilReady(int timeout)
       }//for
       
       tp->unlock_mutex();
-      if (foundAliveNode == theNoOfDBnodes) {
+      if (foundAliveNode == theImpl->theNoOfDBnodes) {
 	DBUG_RETURN(0);
       }//if
       if (foundAliveNode > 0) {
@@ -1077,7 +1066,7 @@ Ndb::guessPrimaryNode(Uint32 fragmentId){
 
 void
 Ndb::StartTransactionNodeSelectionData::init(Uint32 noOfNodes,
-                                             Uint32 nodeIds[]) {
+                                             Uint8 nodeIds[]) {
   kValue           = 6;
   noOfFragments    = 2 * noOfNodes;
 

ndb/src/ndbapi/NdbConnection.cpp

@@ -83,7 +83,7 @@ NdbConnection::NdbConnection( Ndb* aNdb ) :
 {
   theListState = NotInList;
   theError.code = 0;
-  theId = theNdb->theNdbObjectIdMap->map(this);
+  theId = theNdb->theImpl->theNdbObjectIdMap.map(this);
 
 #define CHECK_SZ(mask, sz) assert((sizeof(mask)/sizeof(mask[0])) == sz)
 
@@ -99,7 +99,7 @@ Remark:        Deletes the connection object.
 NdbConnection::~NdbConnection()
 {
   DBUG_ENTER("NdbConnection::~NdbConnection");
-  theNdb->theNdbObjectIdMap->unmap(theId, this);
+  theNdb->theImpl->theNdbObjectIdMap.unmap(theId, this);
   DBUG_VOID_RETURN;
 }//NdbConnection::~NdbConnection()
 

ndb/src/ndbapi/NdbImpl.hpp

@@ -17,7 +17,15 @@
 #ifndef NDB_IMPL_HPP
 #define NDB_IMPL_HPP
 
-#include <Vector.hpp>
+#include <Ndb.hpp>
+#include <NdbError.hpp>
+#include <NdbCondition.h>
+#include <NdbReceiver.hpp>
+#include <NdbOperation.hpp>
+#include <kernel/ndb_limits.h>
+
+#include <NdbTick.h>
+
 #include "ObjectMap.hpp"
 
 /**
@@ -25,19 +33,20 @@
  */
 class NdbImpl {
 public:
-  Vector<class NdbTableImpl *> m_invalidTables;
+  NdbImpl();
+  ~NdbImpl();
 
-  void checkErrorCode(Uint32 i);
-  void checkInvalidTable(class NdbDictionaryImpl * dict);
-};
+  // Ensure good distribution of connects
+  Uint32 theCurrentConnectIndex;
 
-#include <Ndb.hpp>
-#include <NdbError.hpp>
-#include <NdbCondition.h>
-#include <NdbReceiver.hpp>
-#include <NdbOperation.hpp>
+  NdbObjectIdMap theNdbObjectIdMap;
 
-#include <NdbTick.h>
+  Uint32 theNoOfDBnodes; // The number of DB nodes  
+  Uint8 theDBnodes[MAX_NDB_NODES]; // The node number of the DB nodes
+
+ // 1 indicates to release all connections to node 
+  Uint32 the_release_ind[MAX_NDB_NODES];
+};
 
 #ifdef VM_TRACE
 #define TRACE_DEBUG(x) ndbout << x << endl;
@@ -57,7 +66,7 @@ public:
 inline
 void *
 Ndb::int2void(Uint32 val){
-  return theNdbObjectIdMap->getObject(val);
+  return theImpl->theNdbObjectIdMap.getObject(val);
 }
 
 inline

ndb/src/ndbapi/NdbReceiver.cpp

@@ -40,7 +40,7 @@ NdbReceiver::~NdbReceiver()
 {
   DBUG_ENTER("NdbReceiver::~NdbReceiver");
   if (m_id != NdbObjectIdMap::InvalidId) {
-    m_ndb->theNdbObjectIdMap->unmap(m_id, this);
+    m_ndb->theImpl->theNdbObjectIdMap.unmap(m_id, this);
   }
   delete[] m_rows;
   DBUG_VOID_RETURN;
@@ -54,7 +54,7 @@ NdbReceiver::init(ReceiverType type, void* owner)
   m_owner = owner;
   if (m_id == NdbObjectIdMap::InvalidId) {
     if (m_ndb)
-      m_id = m_ndb->theNdbObjectIdMap->map(this);
+      m_id = m_ndb->theImpl->theNdbObjectIdMap.map(this);
   }
 
   theFirstRecAttr = NULL;

ndb/src/ndbapi/Ndbif.cpp

@@ -92,8 +92,8 @@ Ndb::init(int aMaxNoOfTransactions)
   
   theDictionary->setTransporter(this, theFacade);
   
-  aNrOfCon = theNoOfDBnodes;
-  aNrOfOp = 2*theNoOfDBnodes;
+  aNrOfCon = theImpl->theNoOfDBnodes;
+  aNrOfOp = 2*theImpl->theNoOfDBnodes;
   
   // Create connection object in a linked list 
   if((createConIdleList(aNrOfCon)) == -1){
@@ -192,14 +192,14 @@ void Ndb::connected(Uint32 ref)
   }
   
   TransporterFacade * theFacade =  TransporterFacade::instance();
-  int i;
-  theNoOfDBnodes= 0;
+  int i, n= 0;
   for (i = 1; i < MAX_NDB_NODES; i++){
     if (theFacade->getIsDbNode(i)){
-      theDBnodes[theNoOfDBnodes] = i;
-      theNoOfDBnodes++;
+      theImpl->theDBnodes[n] = i;
+      n++;
     }
   }
+  theImpl->theNoOfDBnodes= n;
   theFirstTransId = ((Uint64)tBlockNo << 52)+
     ((Uint64)tmpTheNode << 40);
   theFirstTransId += theFacade->m_max_trans_id;
@@ -207,9 +207,10 @@ void Ndb::connected(Uint32 ref)
   DBUG_PRINT("info",("connected with ref=%x, id=%d, no_db_nodes=%d, first_trans_id=%lx",
 		     theMyRef,
 		     tmpTheNode,
-		     theNoOfDBnodes,
+		     theImpl->theNoOfDBnodes,
 		     theFirstTransId));
-  startTransactionNodeSelectionData.init(theNoOfDBnodes, theDBnodes);
+  startTransactionNodeSelectionData.init(theImpl->theNoOfDBnodes,
+					 theImpl->theDBnodes);
   theCommitAckSignal = new NdbApiSignal(theMyRef);
 
   theDictionary->m_receiver.m_reference= theMyRef;
@@ -247,7 +248,9 @@ Ndb::report_node_failure(Uint32 node_id)
    * 
    * This method is only called by ClusterMgr (via lots of methods)
    */
-  the_release_ind[node_id] = 1;
+  theImpl->the_release_ind[node_id] = 1;
+  // must come after
+  theImpl->the_release_ind[0] = 1;
   theWaiter.nodeFail(node_id);
   return;
 }//Ndb::report_node_failure()

ndb/src/ndbapi/Ndbinit.cpp

@@ -82,7 +82,6 @@ void Ndb::setup(Ndb_cluster_connection *ndb_cluster_connection,
 {
   DBUG_ENTER("Ndb::setup");
 
-  theNdbObjectIdMap= 0;
   m_ndb_cluster_connection= ndb_cluster_connection;
   thePreparedTransactionsArray= NULL;
   theSentTransactionsArray= NULL;
@@ -110,9 +109,6 @@ void Ndb::setup(Ndb_cluster_connection *ndb_cluster_connection,
   theCallList= NULL;
   theScanList= NULL;
   theNdbBlobIdleList= NULL;
-  theNoOfDBnodes= 0;
-  theDBnodes= NULL;
-  the_release_ind= NULL;
   the_last_check_time= 0;
   theFirstTransId= 0;
   theRestartGCI= 0;
@@ -134,19 +130,12 @@ void Ndb::setup(Ndb_cluster_connection *ndb_cluster_connection,
 
   theError.code = 0;
 
-  theNdbObjectIdMap  = new NdbObjectIdMap(1024,1024);
   theConnectionArray = new NdbConnection * [MAX_NDB_NODES];
-  theDBnodes         = new Uint32[MAX_NDB_NODES];
-  the_release_ind    = new Uint8[MAX_NDB_NODES];
   theCommitAckSignal = NULL;
   
-  theCurrentConnectCounter = 1;
-  theCurrentConnectIndex = 0;
   int i;
   for (i = 0; i < MAX_NDB_NODES ; i++) {
     theConnectionArray[i] = NULL;
-    the_release_ind[i] = 0;
-    theDBnodes[i] = 0;
   }//forg
   for (i = 0; i < 2048 ; i++) {
     theFirstTupleId[i] = 0;
@@ -213,7 +202,6 @@ Ndb::~Ndb()
   doDisconnect();
 
   delete theDictionary;  
-  delete theImpl;
 
   NdbGlobalEventBuffer_drop(theGlobalEventBufferHandle);
 
@@ -260,15 +248,12 @@ Ndb::~Ndb()
   startTransactionNodeSelectionData.release();
 
   delete []theConnectionArray;
-  delete []theDBnodes;
-  delete []the_release_ind;
   if(theCommitAckSignal != NULL){
     delete theCommitAckSignal; 
     theCommitAckSignal = NULL;
   }
 
-  if(theNdbObjectIdMap != 0)
-    delete theNdbObjectIdMap;
+  delete theImpl;
 
   /** 
    *  This sleep is to make sure that the transporter 
@@ -307,4 +292,17 @@ NdbWaiter::~NdbWaiter(){
   NdbCondition_Destroy(m_condition);
 }
 
+NdbImpl::NdbImpl() : theNdbObjectIdMap(1024,1024),
+		     theCurrentConnectIndex(0),
+		     theNoOfDBnodes(0)
+{
+  int i;
+  for (i = 0; i < MAX_NDB_NODES; i++) {
+    the_release_ind[i] = 0;
+  }
+}
+
+NdbImpl::~NdbImpl()
+{
+}
 

ndb/src/ndbapi/Ndblist.cpp

@@ -30,10 +30,18 @@ void
 Ndb::checkFailedNode()
 {
   DBUG_ENTER("Ndb::checkFailedNode");
-  DBUG_PRINT("enter", ("theNoOfDBnodes: %d", theNoOfDBnodes));
+  Uint32 *the_release_ind= theImpl->the_release_ind;
+  if (the_release_ind[0] == 0)
+  {
+    DBUG_VOID_RETURN;
+  }
+  Uint32 tNoOfDbNodes = theImpl->theNoOfDBnodes;
+  Uint8 *theDBnodes= theImpl->theDBnodes;
+
+  DBUG_PRINT("enter", ("theNoOfDBnodes: %d", tNoOfDbNodes));
 
-  DBUG_ASSERT(theNoOfDBnodes < MAX_NDB_NODES);
-  for (Uint32 i = 0; i < theNoOfDBnodes; i++){
+  DBUG_ASSERT(tNoOfDbNodes < MAX_NDB_NODES);
+  for (Uint32 i = 0; i < tNoOfDbNodes; i++){
     const NodeId node_id = theDBnodes[i];
     DBUG_PRINT("info", ("i: %d, node_id: %d", i, node_id));
     
@@ -56,31 +64,6 @@ Ndb::checkFailedNode()
   DBUG_VOID_RETURN;
 }
 
-#if 0
-void
-NdbImpl::checkInvalidTable(NdbDictionaryImpl * dict){
-  Uint32 sz = m_invalidTables.size();
-  for(Int32 i = sz - 1; i >= 0; i--){
-    NdbTableImpl * tab = m_invalidTables[i];
-    m_invalidTables.erase(i);
-    dict->tableDropped(* tab);
-  }
-}
-
-void
-NdbImpl::checkErrorCode(Uint32 i, NdbTableImpl * tab){
-  switch(i){
-  case 241:
-  case 283:
-  case 284:
-  case 285:
-  case 1225:
-  case 1226:
-    
-  }
-}
-#endif
-
 /***************************************************************************
  * int createConIdleList(int aNrOfCon);
  * 

sql/log_event.cc

@@ -373,6 +373,9 @@ int Log_event::exec_event(struct st_relay_log_info* rli)
          Note that Rotate_log_event::exec_event() does not call this function,
          so there is no chance that a fake rotate event resets
          last_master_timestamp.
+         Note that we update without mutex (probably ok - except in some very
+         rare cases, only consequence is that value may take some time to
+         display in Seconds_Behind_Master - not critical).
       */
       rli->last_master_timestamp= when;
     }

sql/slave.cc

@@ -545,7 +545,7 @@ int terminate_slave_threads(MASTER_INFO* mi,int thread_mask,bool skip_lock)
 int terminate_slave_thread(THD* thd, pthread_mutex_t* term_lock,
 			   pthread_mutex_t *cond_lock,
 			   pthread_cond_t* term_cond,
-			   volatile bool* slave_running)
+			   volatile uint *slave_running)
 {
   if (term_lock)
   {
@@ -583,7 +583,7 @@ int terminate_slave_thread(THD* thd, pthread_mutex_t* term_lock,
 int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,
 		       pthread_mutex_t *cond_lock,
 		       pthread_cond_t *start_cond,
-		       volatile bool *slave_running,
+		       volatile uint *slave_running,
 		       volatile ulong *slave_run_id,
 		       MASTER_INFO* mi,
                        bool high_priority)
@@ -963,7 +963,7 @@ void end_slave()
 static bool io_slave_killed(THD* thd, MASTER_INFO* mi)
 {
   DBUG_ASSERT(mi->io_thd == thd);
-  DBUG_ASSERT(mi->slave_running == 1); // tracking buffer overrun
+  DBUG_ASSERT(mi->slave_running); // tracking buffer overrun
   return mi->abort_slave || abort_loop || thd->killed;
 }
 
@@ -1767,19 +1767,13 @@ void init_master_info_with_options(MASTER_INFO* mi)
     strmake(mi->ssl_key, master_ssl_key, sizeof(mi->ssl_key)-1);
 }
 
-static void clear_slave_error(RELAY_LOG_INFO* rli)
+void clear_slave_error(RELAY_LOG_INFO* rli)
 {
   /* Clear the errors displayed by SHOW SLAVE STATUS */
   rli->last_slave_error[0]= 0;
   rli->last_slave_errno= 0;
 }
 
-void clear_slave_error_timestamp(RELAY_LOG_INFO* rli)
-{
-  rli->last_master_timestamp= 0;
-  clear_slave_error(rli);
-}
-
 /*
     Reset UNTIL condition for RELAY_LOG_INFO
    SYNOPSYS
@@ -2166,6 +2160,11 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
     String *packet= &thd->packet;
     protocol->prepare_for_resend();
   
+    /*
+      TODO: we read slave_running without run_lock, whereas these variables
+      are updated under run_lock and not data_lock. In 5.0 we should lock
+      run_lock on top of data_lock (with good order).
+    */
     pthread_mutex_lock(&mi->data_lock);
     pthread_mutex_lock(&mi->rli.data_lock);
 
@@ -2226,7 +2225,12 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
     protocol->store(mi->ssl_cipher, &my_charset_bin);
     protocol->store(mi->ssl_key, &my_charset_bin);
 
-    if (mi->rli.last_master_timestamp)
+    /*
+      Seconds_Behind_Master: if SQL thread is running and I/O thread is
+      connected, we can compute it otherwise show NULL (i.e. unknown).
+    */
+    if ((mi->slave_running == MYSQL_SLAVE_RUN_CONNECT) &&
+        mi->rli.slave_running)
     {
       long tmp= (long)((time_t)time((time_t*) 0)
                                - mi->rli.last_master_timestamp)
@@ -2246,9 +2250,13 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
         slave is 2. At SHOW SLAVE STATUS time, assume that the difference
         between timestamp of slave and rli->last_master_timestamp is 0
         (i.e. they are in the same second), then we get 0-(2-1)=-1 as a result.
-        This confuses users, so we don't go below 0.
+        This confuses users, so we don't go below 0: hence the max().
+
+        last_master_timestamp == 0 (an "impossible" timestamp 1970) is a
+        special marker to say "consider we have caught up".
       */
-      protocol->store((longlong)(max(0, tmp)));
+      protocol->store((longlong)(mi->rli.last_master_timestamp ? max(0, tmp)
+                                 : 0));
     }
     else
       protocol->store_null();
@@ -3041,6 +3049,8 @@ extern "C" pthread_handler_decl(handle_slave_io,arg)
 
 connected:
 
+  // TODO: the assignment below should be under mutex (5.0)
+  mi->slave_running= MYSQL_SLAVE_RUN_CONNECT;
   thd->slave_net = &mysql->net;
   thd->proc_info = "Checking master version";
   if (get_master_version_and_clock(mysql, mi))
@@ -3072,6 +3082,7 @@ dump");
 	goto err;
       }
 	  
+      mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT;
       thd->proc_info= "Waiting to reconnect after a failed binlog dump request";
 #ifdef SIGNAL_WITH_VIO_CLOSE
       thd->clear_active_vio();
@@ -3148,6 +3159,7 @@ max_allowed_packet",
 			  mysql_error(mysql));
 	  goto err;
 	}
+        mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT;
 	thd->proc_info = "Waiting to reconnect after a failed master event read";
 #ifdef SIGNAL_WITH_VIO_CLOSE
         thd->clear_active_vio();
@@ -3323,6 +3335,14 @@ extern "C" pthread_handler_decl(handle_slave_sql,arg)
   pthread_mutex_lock(&LOCK_thread_count);
   threads.append(thd);
   pthread_mutex_unlock(&LOCK_thread_count);
+  /*
+    We are going to set slave_running to 1. Assuming slave I/O thread is
+    alive and connected, this is going to make Seconds_Behind_Master be 0
+    i.e. "caught up". Even if we're just at start of thread. Well it's ok, at
+    the moment we start we can think we are caught up, and the next second we
+    start receiving data so we realize we are not caught up and
+    Seconds_Behind_Master grows. No big deal.
+  */
   rli->slave_running = 1;
   rli->abort_slave = 0;
   pthread_mutex_unlock(&rli->run_lock);
@@ -4169,6 +4189,21 @@ Before assert, my_b_tell(cur_log)=%s  rli->event_relay_log_pos=%s",
       */
       if (hot_log)
       {
+        /*
+          We say in Seconds_Behind_Master that we have "caught up". Note that
+          for example if network link is broken but I/O slave thread hasn't
+          noticed it (slave_net_timeout not elapsed), then we'll say "caught
+          up" whereas we're not really caught up. Fixing that would require
+          internally cutting timeout in smaller pieces in network read, no
+          thanks. Another example: SQL has caught up on I/O, now I/O has read
+          a new event and is queuing it; the false "0" will exist until SQL
+          finishes executing the new event; it will be look abnormal only if
+          the events have old timestamps (then you get "many", 0, "many").
+          Transient phases like this can't really be fixed.
+        */
+        time_t save_timestamp= rli->last_master_timestamp;
+        rli->last_master_timestamp= 0;
+
 	DBUG_ASSERT(rli->relay_log.get_open_count() == rli->cur_log_old_open_count);
 	/*
 	  We can, and should release data_lock while we are waiting for
@@ -4215,6 +4250,7 @@ Before assert, my_b_tell(cur_log)=%s  rli->event_relay_log_pos=%s",
         rli->relay_log.wait_for_update(rli->sql_thd, 1);
         // re-acquire data lock since we released it earlier
         pthread_mutex_lock(&rli->data_lock);
+        rli->last_master_timestamp= save_timestamp;
 	continue;
       }
       /*

sql/slave.h

@@ -98,6 +98,21 @@ enum enum_binlog_formats {
   BINLOG_FORMAT_323_LESS_57, 
   BINLOG_FORMAT_323_GEQ_57 };
 
+/*
+  3 possible values for MASTER_INFO::slave_running and
+  RELAY_LOG_INFO::slave_running.
+  The values 0,1,2 are very important: to keep the diff small, I didn't
+  substitute places where we use 0/1 with the newly defined symbols. So don't change
+  these values.
+  The same way, code is assuming that in RELAY_LOG_INFO we use only values
+  0/1.
+  I started with using an enum, but
+  enum_variable=1; is not legal so would have required many line changes.
+*/
+#define MYSQL_SLAVE_NOT_RUN         0
+#define MYSQL_SLAVE_RUN_NOT_CONNECT 1
+#define MYSQL_SLAVE_RUN_CONNECT     2
+
 /****************************************************************************
 
   Replication SQL Thread
@@ -251,7 +266,8 @@ typedef struct st_relay_log_info
 
   /* if not set, the value of other members of the structure are undefined */
   bool inited;
-  volatile bool abort_slave, slave_running;
+  volatile bool abort_slave;
+  volatile uint slave_running;
 
   /* 
      Condition and its parameters from START SLAVE UNTIL clause.
@@ -385,7 +401,8 @@ typedef struct st_master_info
 #endif
   bool inited;
   enum enum_binlog_formats old_format;
-  volatile bool abort_slave, slave_running;
+  volatile bool abort_slave;
+  volatile uint slave_running;
   volatile ulong slave_run_id;
   /* 
      The difference in seconds between the clock of the master and the clock of
@@ -464,7 +481,7 @@ int terminate_slave_threads(MASTER_INFO* mi, int thread_mask,
 int terminate_slave_thread(THD* thd, pthread_mutex_t* term_mutex,
 			   pthread_mutex_t* cond_lock,
 			   pthread_cond_t* term_cond,
-			   volatile bool* slave_running);
+			   volatile uint* slave_running);
 int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
 			MASTER_INFO* mi, const char* master_info_fname,
 			const char* slave_info_fname, int thread_mask);
@@ -477,7 +494,7 @@ int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
 int start_slave_thread(pthread_handler h_func, pthread_mutex_t* start_lock,
 		       pthread_mutex_t *cond_lock,
 		       pthread_cond_t* start_cond,
-		       volatile bool *slave_running,
+		       volatile uint *slave_running,
 		       volatile ulong *slave_run_id,
 		       MASTER_INFO* mi,
                        bool high_priority);
@@ -519,7 +536,7 @@ void slave_print_error(RELAY_LOG_INFO* rli, int err_code, const char* msg, ...);
 void end_slave(); /* clean up */
 void init_master_info_with_options(MASTER_INFO* mi);
 void clear_until_condition(RELAY_LOG_INFO* rli);
-void clear_slave_error_timestamp(RELAY_LOG_INFO* rli);
+void clear_slave_error(RELAY_LOG_INFO* rli);
 int init_master_info(MASTER_INFO* mi, const char* master_info_fname,
 		     const char* slave_info_fname,
 		     bool abort_if_no_master_info_file,

sql/sql_repl.cc

@@ -880,10 +880,10 @@ int reset_slave(THD *thd, MASTER_INFO* mi)
   */
   init_master_info_with_options(mi);
   /* 
-     Reset errors, and master timestamp (the idea is that we forget about the
+     Reset errors (the idea is that we forget about the
      old master).
   */
-  clear_slave_error_timestamp(&mi->rli);
+  clear_slave_error(&mi->rli);
   clear_until_condition(&mi->rli);
   
   // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
@@ -1143,8 +1143,8 @@ int change_master(THD* thd, MASTER_INFO* mi)
 
   pthread_mutex_lock(&mi->rli.data_lock);
   mi->rli.abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
-  /* Clear the errors, for a clean start, and master timestamp */
-  clear_slave_error_timestamp(&mi->rli);
+  /* Clear the errors, for a clean start */
+  clear_slave_error(&mi->rli);
   clear_until_condition(&mi->rli);
   /*
     If we don't write new coordinates to disk now, then old will remain in