Merge mysql.com:/home/gluh/MySQL/Merge/5.0

into  mysql.com:/home/gluh/MySQL/Merge/5.0-opt

mysql-test/r/ndb_basic.result

@@ -749,3 +749,19 @@ f1	f2	f3
 222222	bbbbbb	2
 drop table t1;
 Illegal ndb error code: 1186
+CREATE TABLE t1 (
+a VARBINARY(40) NOT NULL,
+b VARCHAR (256) CHARACTER SET UTF8 NOT NULL,
+c VARCHAR(256) CHARACTER SET UTF8 NOT NULL,
+PRIMARY KEY (b,c))  ENGINE=ndbcluster;
+INSERT INTO t1 VALUES
+("a","ab","abc"),("b","abc","abcd"),("c","abc","ab"),("d","ab","ab"),("e","abc","abc");
+SELECT * FROM t1 ORDER BY a;
+a	b	c
+a	ab	abc
+b	abc	abcd
+c	abc	ab
+d	ab	ab
+e	abc	abc
+DROP TABLE t1;
+End of 5.0 tests

mysql-test/r/ndb_index_ordered.result

@@ -658,3 +658,11 @@ insert into t1 (a, c) values (1,'aaa'),(3,'bbb');
 select count(*) from t1 where c<'bbb';
 count(*)
 1
+create table nationaldish (DishID int(10) unsigned NOT NULL AUTO_INCREMENT,
+CountryCode char(3) NOT NULL,
+DishTitle varchar(64) NOT NULL,
+calories smallint(5) unsigned DEFAULT NULL,
+PRIMARY KEY (DishID),
+INDEX i USING HASH (countrycode,calories)
+) ENGINE=ndbcluster;
+ERROR HY000: Can't create table './test/nationaldish.frm' (errno: 138)

mysql-test/r/ndb_read_multi_range.result

@@ -367,3 +367,17 @@ a	b	c
 406994	67	2006-02-27 11:26:46
 406995	67	2006-02-28 11:55:00
 DROP TABLE t1, t11, t12, t21, t22;
+CREATE TABLE t1 (id varchar(255) NOT NULL,
+tag int(11) NOT NULL,
+doc text NOT NULL,
+type varchar(150) NOT NULL,
+modified timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+PRIMARY KEY (id)
+) ENGINE=ndbcluster;
+INSERT INTO t1 VALUES ('sakila',1,'Some text goes here','text',CURRENT_TIMESTAMP);
+SELECT id, tag, doc, type FROM t1 WHERE id IN ('flipper','orka');
+id	tag	doc	type
+SELECT id, tag, doc, type FROM t1 WHERE id IN ('flipper','sakila');
+id	tag	doc	type
+sakila	1	Some text goes here	text
+DROP TABLE t1;

mysql-test/t/ndb_basic.test

@@ -710,3 +710,22 @@ drop table t1;
 --error 1
 --exec $MY_PERROR --ndb 1186 2>&1
 
+#
+# Bug #25746 - VARCHAR UTF8 PK issue
+# - prior to bugfix 4209, illegal length parameter would be 
+# returned in SELECT *
+
+CREATE TABLE t1 (
+a VARBINARY(40) NOT NULL,
+b VARCHAR (256) CHARACTER SET UTF8 NOT NULL,
+c VARCHAR(256) CHARACTER SET UTF8 NOT NULL,
+PRIMARY KEY (b,c))  ENGINE=ndbcluster;
+INSERT INTO t1 VALUES
+("a","ab","abc"),("b","abc","abcd"),("c","abc","ab"),("d","ab","ab"),("e","abc","abc");
+SELECT * FROM t1 ORDER BY a;
+DROP TABLE t1;
+
+# End of 5.0 tests
+--echo End of 5.0 tests
+
+

mysql-test/t/ndb_index_ordered.test

@@ -356,3 +356,15 @@ insert into t1 (a, c) values (1,'aaa'),(3,'bbb');
 select count(*) from t1 where c<'bbb';
 
 # End of 4.1 tests
+
+# bug#24820 CREATE INDEX ....USING HASH on NDB table creates ordered index, not HASH index
+
+--error ER_CANT_CREATE_TABLE
+create table nationaldish (DishID int(10) unsigned NOT NULL AUTO_INCREMENT,
+                           CountryCode char(3) NOT NULL,
+                           DishTitle varchar(64) NOT NULL,
+                           calories smallint(5) unsigned DEFAULT NULL,
+                           PRIMARY KEY (DishID),
+                           INDEX i USING HASH (countrycode,calories)
+                           ) ENGINE=ndbcluster;
+

mysql-test/t/ndb_read_multi_range.test

@@ -238,3 +238,18 @@ select * from t12 order by 1,2,3;
 select * from t21 order by 1,2,3;
 select * from t22 order by 1,2,3;
 DROP TABLE t1, t11, t12, t21, t22;
+
+# bug#19956
+CREATE TABLE t1 (id varchar(255) NOT NULL,
+		 tag int(11) NOT NULL,
+                 doc text NOT NULL,
+                 type varchar(150) NOT NULL,
+                 modified timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                 PRIMARY KEY (id)
+                ) ENGINE=ndbcluster;
+
+INSERT INTO t1 VALUES ('sakila',1,'Some text goes here','text',CURRENT_TIMESTAMP);
+SELECT id, tag, doc, type FROM t1 WHERE id IN ('flipper','orka');
+SELECT id, tag, doc, type FROM t1 WHERE id IN ('flipper','sakila');
+
+DROP TABLE t1;

ndb/include/kernel/signaldata/DumpStateOrd.hpp

@@ -67,6 +67,7 @@ public:
     // 100-105 TUP and ACC  
     // 200-240 UTIL
     // 300-305 TRIX
+    QmgrErr935 = 935,
     NdbfsDumpFileStat = 400,
     NdbfsDumpAllFiles = 401,
     NdbfsDumpOpenFiles = 402,

ndb/include/mgmapi/mgmapi.h

@@ -540,6 +540,16 @@ extern "C" {
    */
   const char *ndb_mgm_get_connectstring(NdbMgmHandle handle, char *buf, int buf_sz);
 
+  /**
+   * Sets the number of seconds to wait for connect(2) during ndb_mgm_connect
+   * Default is no timeout
+   *
+   * @param handle  NdbMgmHandle
+   * @param seconds number of seconds
+   * @return non-zero on success
+   */
+  int ndb_mgm_set_connect_timeout(NdbMgmHandle handle, unsigned int seconds);
+
   /**
    * Connects to a management server. Connectstring is set by
    * ndb_mgm_set_connectstring().

ndb/include/portlib/NdbMem.h

@@ -65,7 +65,7 @@ void NdbMem_Free(void* ptr);
  * NdbMem_MemLockAll
  *   Locks virtual memory in main memory
  */
-int NdbMem_MemLockAll(void);
+int NdbMem_MemLockAll(int);
 
 /**
  * NdbMem_MemUnlockAll

ndb/include/util/SocketClient.hpp

@@ -23,6 +23,7 @@ class SocketClient
 {
   NDB_SOCKET_TYPE m_sockfd;
   struct sockaddr_in m_servaddr;
+  unsigned int m_connect_timeout_sec;
   unsigned short m_port;
   char *m_server_name;
   SocketAuthenticator *m_auth;
@@ -34,6 +35,9 @@ public:
     m_port = port;
     m_servaddr.sin_port = htons(m_port);
   };
+  void set_connect_timeout(unsigned int s) {
+    m_connect_timeout_sec= s;
+  }
   unsigned short get_port() { return m_port; };
   char *get_server_name() { return m_server_name; };
   int bind(const char* toaddress, unsigned short toport);

ndb/src/common/debugger/EventLogger.cpp

@@ -115,7 +115,8 @@ void getTextNDBStopForced(QQQQ) {
   int sphase        = theData[4];
   int extra         = theData[5];
   getRestartAction(theData[1],action_str);
-  reason_str.appfmt(" Initiated by signal %d.", signum);
+  if (signum)
+    reason_str.appfmt(" Initiated by signal %d.", signum);
   if (error)
   {
     ndbd_exit_classification cl;

ndb/src/common/portlib/NdbMem.c

@@ -56,7 +56,15 @@ void NdbMem_Free(void* ptr)
 }
 
  
-int NdbMem_MemLockAll(){
+int NdbMem_MemLockAll(int i){
+  if (i == 1)
+  {
+#if defined(HAVE_MLOCKALL) && defined(MCL_CURRENT) && defined (MCL_FUTURE)
+    return mlockall(MCL_CURRENT | MCL_FUTURE);
+#else
+    return -1;
+#endif
+  }
 #if defined(HAVE_MLOCKALL) && defined(MCL_CURRENT)
   return mlockall(MCL_CURRENT);
 #else

ndb/src/common/transporter/Transporter.cpp

@@ -79,9 +79,13 @@ Transporter::Transporter(TransporterRegistry &t_reg,
   if (isServer)
     m_socket_client= 0;
   else
+  {
     m_socket_client= new SocketClient(remoteHostName, s_port,
 				      new SocketAuthSimple("ndbd",
 							   "ndbd passwd"));
+
+    m_socket_client->set_connect_timeout((m_timeOutMillis+999)/1000);
+  }
   DBUG_VOID_RETURN;
 }
 
@@ -140,9 +144,9 @@ Transporter::connect_client() {
     }
     sockfd= m_socket_client->connect();
   }
-  
+
   return connect_client(sockfd);
-}  
+}
 
 bool
 Transporter::connect_client(NDB_SOCKET_TYPE sockfd) {

ndb/src/common/util/ConfigValues.cpp

@@ -48,7 +48,7 @@ static const char Magic[] = { 'N', 'D', 'B', 'C', 'O', 'N', 'F', 'V' };
 
 //#define DEBUG_CV
 #ifdef DEBUG_CV
-#define DEBUG
+#define DEBUG if(getenv("CV_DEBUG"))
 #else
 #define DEBUG if(0)
 #endif
@@ -216,62 +216,60 @@ ConfigValues::Iterator::set(Uint32 key, const char * value){
 static
 bool
 findKey(const Uint32 * values, Uint32 sz, Uint32 key, Uint32 * _pos){
-  Uint32 pos = hash(key, sz);
-  Uint32 count = 0;
-  while((values[pos] & KP_MASK) != key && count < sz){
-    pos = nextHash(key, sz, pos, ++count);
-  }
+  Uint32 lo = 0;
+  Uint32 hi = sz;
+  Uint32 pos = (hi + lo) >> 1;
 
-  if((values[pos] & KP_MASK)== key){
-    *_pos = pos;
-    return true;
+  DEBUG printf("findKey(H'%.8x %d)", key, sz);
+
+  if (sz == 0)
+  {
+    DEBUG ndbout_c(" -> false, 0");
+    * _pos = 0;
+    return false;
   }
-  return false;
-}
 
-static
-Uint32
-hash(Uint32 key, Uint32 size){
-  Uint32 tmp = (key >> 16) ^ (key & 0xFFFF);
-  return (((tmp << 16) | tmp) % size) << 1;
-}
+  Uint32 val = 0;
+  Uint32 oldpos = pos + 1;
+  while (pos != oldpos) 
+  {
+    DEBUG printf(" [ %d %d %d ] ", lo, pos, hi);
+    assert(pos < hi);
+    assert(pos >= lo);
+    val = values[2*pos] & KP_MASK;
+    if (key > val)
+    {
+      lo = pos;
+    }
+    else if (key < val)
+    {
+      hi = pos;
+    }
+    else 
+    {
+      * _pos = 2*pos;
+      DEBUG ndbout_c(" -> true, %d", pos);
+      return true;
+    }
+    oldpos = pos;
+    pos = (hi + lo) >> 1;
+  } 
 
-static
-Uint32
-nextHash(Uint32 key, Uint32 size, Uint32 pos, Uint32 count){
-  Uint32 p = (pos >> 1);
-  if((key % size) != 0)
-    p += key;
-  else
-    p += 1;
-  return (p % size) << 1;
-}
+  DEBUG printf(" pos: %d (key %.8x val: %.8x values[pos]: %x) key>val: %d ",
+	       pos, key, val, values[2*pos] & KP_MASK,
+	       key > val);
 
-static
-Uint32
-directory(Uint32 sz){
-  const Uint32 _input = sz;
-  if((sz & 1) == 0)
-    sz ++;
+  pos += (key > val) ? 1 : 0;
   
-  bool prime = false;
-  while(!prime){
-    prime = true;
-    for(Uint32 n = 3; n*n <= sz; n += 2){
-      if((sz % n) == 0){
-	prime = false;
-	sz += 2;
-	break;
-      }
-    }
-  }
-  DEBUG printf("directory %d -> %d\n", _input, sz);
-  return sz;
+  * _pos = 2*pos;
+  DEBUG ndbout_c(" -> false, %d", pos);
+  return false;
 }
 
+
 ConfigValuesFactory::ConfigValuesFactory(Uint32 keys, Uint32 data){
   m_sectionCounter = (1 << KP_SECTION_SHIFT);
-  m_freeKeys = directory(keys);
+  m_freeKeys = keys;
   m_freeData = (data + 7) & ~7;
   m_currentSection = 0;
   m_cfg = create(m_freeKeys, m_freeData);
@@ -330,11 +328,14 @@ ConfigValuesFactory::expand(Uint32 fk, Uint32 fs){
     return ;
   }
 
+  DEBUG printf("[ fk fd ] : [ %d %d ]", m_freeKeys, m_freeData);
+  
   m_freeKeys = (m_freeKeys >= fk ? m_cfg->m_size : fk + m_cfg->m_size);
   m_freeData = (m_freeData >= fs ? m_cfg->m_dataSize : fs + m_cfg->m_dataSize);
-  m_freeKeys = directory(m_freeKeys);
   m_freeData = (m_freeData + 7) & ~7;
- 
+  
+  DEBUG ndbout_c(" [ %d %d ]", m_freeKeys, m_freeData);
+
   ConfigValues * m_tmp = m_cfg;
   m_cfg = create(m_freeKeys, m_freeData);
   put(* m_tmp);
@@ -350,7 +351,6 @@ ConfigValuesFactory::shrink(){
 
   m_freeKeys = m_cfg->m_size - m_freeKeys;
   m_freeData = m_cfg->m_dataSize - m_freeData;
-  m_freeKeys = directory(m_freeKeys);
   m_freeData = (m_freeData + 7) & ~7;
 
   ConfigValues * m_tmp = m_cfg;
@@ -429,52 +429,58 @@ ConfigValuesFactory::put(const ConfigValues::Entry & entry){
   }
   
   const Uint32 tmp = entry.m_key | m_currentSection;
-  const Uint32 sz = m_cfg->m_size;
-  Uint32 pos = hash(tmp, sz);
-  Uint32 count = 0;
-  Uint32 val = m_cfg->m_values[pos];
-
-  while((val & KP_MASK) != tmp && val != CFV_KEY_FREE && count < sz){
-    pos = nextHash(tmp, sz, pos, ++count);
-    val = m_cfg->m_values[pos];
-  }
+  const Uint32 sz = m_cfg->m_size - m_freeKeys;
 
-  if((val & KP_MASK) == tmp){
+  Uint32 pos;
+  if (findKey(m_cfg->m_values, sz, tmp, &pos))
+  {
     DEBUG ndbout_c("key %x already found at pos: %d", tmp, pos);
     return false;
   }
 
-  if(count >= sz){
-    pos = hash(tmp, sz);    
-    count = 0;
-    Uint32 val = m_cfg->m_values[pos];
-   
-    printf("key: %d, (key %% size): %d\n", entry.m_key, (entry.m_key % sz));
-    printf("pos: %d", pos);
-    while((val & KP_MASK) != tmp && val != CFV_KEY_FREE && count < sz){
-      pos = nextHash(tmp, sz, pos, ++count);
-      val = m_cfg->m_values[pos];
-      printf(" %d", pos);
+  DEBUG {
+    printf("H'before ");
+    Uint32 prev = 0;
+    for (Uint32 i = 0; i<sz; i++)
+    {
+      Uint32 val = m_cfg->m_values[2*i] & KP_MASK;
+      ndbout_c("%.8x", val);
+      assert(val >= prev);
+      prev = val;
     }
-    printf("\n");
-
-    abort();
-    printf("Full\n");
-    return false;
+  }
+  
+  if (pos != 2*sz)
+  {
+    DEBUG ndbout_c("pos: %d sz: %d", pos, sz);
+    memmove(m_cfg->m_values + pos + 2, m_cfg->m_values + pos, 
+	    4 * (2*sz - pos));
   }
 
-  assert(pos < (sz << 1));
 
   Uint32 key = tmp;
   key |= (entry.m_type << KP_TYPE_SHIFT);
   m_cfg->m_values[pos] = key;
+
+  DEBUG {
+    printf("H'after ");
+    Uint32 prev = 0;
+    for (Uint32 i = 0; i<=sz; i++)
+    {
+      Uint32 val = m_cfg->m_values[2*i] & KP_MASK;
+      ndbout_c("%.8x", val);
+      assert(val >= prev);
+      prev = val;
+    }
+  }
+  
   switch(entry.m_type){
   case ConfigValues::IntType:
   case ConfigValues::SectionType:
     m_cfg->m_values[pos+1] = entry.m_int;    
     m_freeKeys--;
     DEBUG printf("Putting at: %d(%d) (loop = %d) key: %d value: %d\n", 
-		   pos, sz, count, 
+		   pos, sz, 0, 
 		   (key >> KP_KEYVAL_SHIFT) & KP_KEYVAL_MASK,
 		   entry.m_int);
     return true;
@@ -486,7 +492,7 @@ ConfigValuesFactory::put(const ConfigValues::Entry & entry){
     m_freeKeys--;
     m_freeData -= sizeof(char *);
     DEBUG printf("Putting at: %d(%d) (loop = %d) key: %d value(%d): %s\n", 
-		   pos, sz, count, 
+		   pos, sz, 0, 
 		   (key >> KP_KEYVAL_SHIFT) & KP_KEYVAL_MASK,
 		   index,
 		   entry.m_string);
@@ -499,7 +505,7 @@ ConfigValuesFactory::put(const ConfigValues::Entry & entry){
     m_freeKeys--;
     m_freeData -= 8;
     DEBUG printf("Putting at: %d(%d) (loop = %d) key: %d value64(%d): %lld\n", 
-		   pos, sz, count, 
+		   pos, sz, 0, 
 		   (key >> KP_KEYVAL_SHIFT) & KP_KEYVAL_MASK,
 		   index,
 		   entry.m_int64);
@@ -662,7 +668,9 @@ ConfigValuesFactory::unpack(const void * _src, Uint32 len){
   }
 
   const char * src = (const char *)_src;
-
+  const char * end = src + len - 4;
+  src += sizeof(Magic);
+  
   {
     Uint32 len32 = (len >> 2);
     const Uint32 * tmp = (const Uint32*)_src;
@@ -677,9 +685,37 @@ ConfigValuesFactory::unpack(const void * _src, Uint32 len){
     }
   }
 
-  const char * end = src + len - 4;
-  src += sizeof(Magic);
- 
+  const char * save = src;
+
+  {
+    Uint32 keys = 0;
+    Uint32 data = 0;
+    while(end - src > 4){
+      Uint32 tmp = ntohl(* (const Uint32 *)src); src += 4;
+      keys++;
+      switch(::getTypeOf(tmp)){
+      case ConfigValues::IntType:
+      case ConfigValues::SectionType:
+	src += 4;
+	break;
+      case ConfigValues::Int64Type:
+	src += 8;
+	data += 8;
+	break;
+      case ConfigValues::StringType:{
+	Uint32 s_len = ntohl(* (const Uint32 *)src);
+	src += 4 + mod4(s_len);
+	data += sizeof(char*);
+	break;
+      }
+      default:
+	break;
+      }
+    }
+    expand(keys, data);
+  }
+  
+  src = save;
   ConfigValues::Entry entry;
   while(end - src > 4){
     Uint32 tmp = ntohl(* (const Uint32 *)src); src += 4;

ndb/src/common/util/SocketClient.cpp

@@ -26,6 +26,7 @@ SocketClient::SocketClient(const char *server_name, unsigned short port, SocketA
   m_port= port;
   m_server_name= server_name ? strdup(server_name) : 0;
   m_sockfd= NDB_INVALID_SOCKET;
+  m_connect_timeout_sec= 0;
 }
 
 SocketClient::~SocketClient()
@@ -58,7 +59,7 @@ SocketClient::init()
   if (m_sockfd == NDB_INVALID_SOCKET) {
     return false;
   }
-  
+
   DBUG_PRINT("info",("NDB_SOCKET: %d", m_sockfd));
 
   return true;
@@ -104,6 +105,13 @@ SocketClient::bind(const char* bindaddress, unsigned short localport)
 NDB_SOCKET_TYPE
 SocketClient::connect(const char *toaddress, unsigned short toport)
 {
+  fd_set rset, wset;
+  struct timeval tval;
+  int r;
+  bool use_timeout;
+  SOCKOPT_OPTLEN_TYPE len;
+  int flags;
+
   if (m_sockfd == NDB_INVALID_SOCKET)
   {
     if (!init()) {
@@ -127,14 +135,58 @@ SocketClient::connect(const char *toaddress, unsigned short toport)
     if (Ndb_getInAddr(&m_servaddr.sin_addr, m_server_name))
       return NDB_INVALID_SOCKET;
   }
-  
-  const int r = ::connect(m_sockfd, (struct sockaddr*) &m_servaddr, sizeof(m_servaddr));
-  if (r == -1) {
+
+  flags= fcntl(m_sockfd, F_GETFL, 0);
+  fcntl(m_sockfd, F_SETFL, flags | O_NONBLOCK);
+
+  r= ::connect(m_sockfd, (struct sockaddr*) &m_servaddr, sizeof(m_servaddr));
+
+  if (r == 0)
+    goto done; // connected immediately.
+
+  if (r < 0 && (errno != EINPROGRESS)) {
+    NDB_CLOSE_SOCKET(m_sockfd);
+    m_sockfd= NDB_INVALID_SOCKET;
+    return NDB_INVALID_SOCKET;
+  }
+
+  FD_ZERO(&rset);
+  FD_SET(m_sockfd, &rset);
+  wset= rset;
+  tval.tv_sec= m_connect_timeout_sec;
+  tval.tv_usec= 0;
+  use_timeout= m_connect_timeout_sec;
+
+  if ((r= select(m_sockfd+1, &rset, &wset, NULL,
+                 use_timeout? &tval : NULL)) == 0)
+  {
     NDB_CLOSE_SOCKET(m_sockfd);
     m_sockfd= NDB_INVALID_SOCKET;
     return NDB_INVALID_SOCKET;
   }
 
+  if (FD_ISSET(m_sockfd, &rset) || FD_ISSET(m_sockfd, &wset))
+  {
+    len= sizeof(r);
+    if (getsockopt(m_sockfd, SOL_SOCKET, SO_ERROR, &r, &len) < 0 || r)
+    {
+      // Solaris got an error... different than others
+      NDB_CLOSE_SOCKET(m_sockfd);
+      m_sockfd= NDB_INVALID_SOCKET;
+      return NDB_INVALID_SOCKET;
+    }
+  }
+  else
+  {
+    // select error, probably m_sockfd not set.
+    NDB_CLOSE_SOCKET(m_sockfd);
+    m_sockfd= NDB_INVALID_SOCKET;
+    return NDB_INVALID_SOCKET;
+  }
+
+done:
+  fcntl(m_sockfd, F_SETFL, flags);
+
   if (m_auth) {
     if (!m_auth->client_authenticate(m_sockfd))
     {

ndb/src/kernel/blocks/ERROR_codes.txt

@@ -21,6 +21,9 @@ Crash president when he starts to run in ArbitState 1-9.
 
 910: Crash new president after node crash
 
+935 : Crash master on node failure (delayed) 
+      and skip sending GSN_COMMIT_FAILREQ to specified node
+
 ERROR CODES FOR TESTING NODE FAILURE, GLOBAL CHECKPOINT HANDLING:
 -----------------------------------------------------------------
 

ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp

@@ -341,9 +341,9 @@ void Cmvmi::execSTTOR(Signal* signal)
   if (theStartPhase == 1){
     jam();
 
-    if(theConfig.lockPagesInMainMemory())
+    if(theConfig.lockPagesInMainMemory() == 1)
     {
-      int res = NdbMem_MemLockAll();
+      int res = NdbMem_MemLockAll(0);
       if(res != 0){
 	g_eventLogger.warning("Failed to memlock pages");
 	warningEvent("Failed to memlock pages");
@@ -788,6 +788,21 @@ Cmvmi::execSTART_ORD(Signal* signal) {
   
   if(globalData.theStartLevel == NodeState::SL_CMVMI){
     jam();
+
+    if(theConfig.lockPagesInMainMemory() == 2)
+    {
+      int res = NdbMem_MemLockAll(1);
+      if(res != 0)
+      {
+	g_eventLogger.warning("Failed to memlock pages");
+	warningEvent("Failed to memlock pages");
+      }
+      else
+      {
+	g_eventLogger.info("Locked future allocations");
+      }
+    }
+    
     globalData.theStartLevel  = NodeState::SL_STARTING;
     globalData.theRestartFlag = system_started;
     /**

ndb/src/kernel/blocks/dbdih/DbdihMain.cpp

@@ -3554,7 +3554,6 @@ void Dbdih::endTakeOver(Uint32 takeOverPtrI)
   takeOverPtr.i = takeOverPtrI;
   ptrCheckGuard(takeOverPtr, MAX_NDB_NODES, takeOverRecord);
 
-  releaseTakeOver(takeOverPtrI);
   if ((takeOverPtr.p->toMasterStatus != TakeOverRecord::IDLE) &&
       (takeOverPtr.p->toMasterStatus != TakeOverRecord::TO_WAIT_START_TAKE_OVER)) {
     jam();
@@ -3568,6 +3567,7 @@ void Dbdih::endTakeOver(Uint32 takeOverPtrI)
   }//if
   setAllowNodeStart(takeOverPtr.p->toStartingNode, true);
   initTakeOver(takeOverPtr);
+  releaseTakeOver(takeOverPtrI);
 }//Dbdih::endTakeOver()
 
 void Dbdih::releaseTakeOver(Uint32 takeOverPtrI)
@@ -4709,6 +4709,7 @@ void Dbdih::handleTakeOverNewMaster(Signal* signal, Uint32 takeOverPtrI)
       break;
     }
     ndbrequire(ok);
+    endTakeOver(takeOverPtr.i);
   }//if
 }//Dbdih::handleTakeOverNewMaster()
 

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

@@ -138,7 +138,6 @@
 
 #define ZNOT_FOUND 626
 #define ZALREADYEXIST 630
-#define ZINCONSISTENTHASHINDEX 892
 #define ZNOTUNIQUE 893
 
 #define ZINVALID_KEY 290

ndb/src/kernel/blocks/qmgr/Qmgr.hpp

@@ -425,6 +425,10 @@ private:
   
   StopReq c_stopReq;
   bool check_multi_node_shutdown(Signal* signal);
+
+#ifdef ERROR_INSERT
+  Uint32 c_error_insert_extra;
+#endif
 };
 
 #endif

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

@@ -2894,6 +2894,17 @@ void Qmgr::failReportLab(Signal* signal, Uint16 aFailedNode,
     systemErrorLab(signal, __LINE__);
     return;
   }//if
+
+  if (getNodeState().startLevel < NodeState::SL_STARTED)
+  {
+    jam();
+    CRASH_INSERTION(932);
+    char buf[100];
+    BaseString::snprintf(buf, 100, "Node failure during restart");
+    progError(__LINE__, NDBD_EXIT_SR_OTHERNODEFAILED, buf);
+    ndbrequire(false);
+  }
+
   TnoFailedNodes = cnoFailedNodes;
   failReport(signal, failedNodePtr.i, (UintR)ZTRUE, aFailCause);
   if (cpresident == getOwnNodeId()) {
@@ -2980,6 +2991,16 @@ void Qmgr::execPREP_FAILREQ(Signal* signal)
     return;
   }//if
 
+  if (getNodeState().startLevel < NodeState::SL_STARTED)
+  {
+    jam();
+    CRASH_INSERTION(932);
+    char buf[100];
+    BaseString::snprintf(buf, 100, "Node failure during restart");
+    progError(__LINE__, NDBD_EXIT_SR_OTHERNODEFAILED, buf);
+    ndbrequire(false);
+  }
+
   guard0 = cnoPrepFailedNodes - 1;
   arrGuard(guard0, MAX_NDB_NODES);
   for (Tindex = 0; Tindex <= guard0; Tindex++) {
@@ -3157,6 +3178,18 @@ Qmgr::sendCommitFailReq(Signal* signal)
   for (nodePtr.i = 1; nodePtr.i < MAX_NDB_NODES; nodePtr.i++) {
     jam();
     ptrAss(nodePtr, nodeRec);
+
+#ifdef ERROR_INSERT    
+    if (ERROR_INSERTED(935) && nodePtr.i == c_error_insert_extra)
+    {
+      ndbout_c("skipping node %d", c_error_insert_extra);
+      CLEAR_ERROR_INSERT_VALUE;
+      signal->theData[0] = 9999;
+      sendSignalWithDelay(CMVMI_REF, GSN_NDB_TAMPER, signal, 1000, 1);
+      continue;
+    }
+#endif
+
     if (nodePtr.p->phase == ZRUNNING) {
       jam();
       nodePtr.p->sendCommitFailReqStatus = Q_ACTIVE;
@@ -3227,6 +3260,33 @@ void Qmgr::execPREP_FAILREF(Signal* signal)
   return;
 }//Qmgr::execPREP_FAILREF()
 
+static
+Uint32
+clear_nodes(Uint32 dstcnt, Uint16 dst[], Uint32 srccnt, const Uint16 src[])
+{
+  if (srccnt == 0)
+    return dstcnt;
+  
+  Uint32 pos = 0;
+  for (Uint32 i = 0; i<dstcnt; i++)
+  {
+    Uint32 node = dst[i];
+    for (Uint32 j = 0; j<srccnt; j++)
+    {
+      if (node == dst[j])
+      {
+	node = RNIL;
+	break;
+      }
+    }
+    if (node != RNIL)
+    {
+      dst[pos++] = node;
+    }
+  }
+  return pos;
+}
+
 /*---------------------------------------------------------------------------*/
 /*    THE PRESIDENT IS NOW COMMITTING THE PREVIOUSLY PREPARED NODE FAILURE.  */
 /*---------------------------------------------------------------------------*/
@@ -3314,19 +3374,18 @@ void Qmgr::execCOMMIT_FAILREQ(Signal* signal)
 		   NodeFailRep::SignalLength, JBB);
       }//if
     }//for
-    if (cpresident != getOwnNodeId()) {
-      jam();
-      cnoFailedNodes = cnoCommitFailedNodes - cnoFailedNodes;
-      if (cnoFailedNodes > 0) {
-        jam();
-        guard0 = cnoFailedNodes - 1;
-        arrGuard(guard0 + cnoCommitFailedNodes, MAX_NDB_NODES);
-        for (Tj = 0; Tj <= guard0; Tj++) {
-          jam();
-          cfailedNodes[Tj] = cfailedNodes[Tj + cnoCommitFailedNodes];
-        }//for
-      }//if
-    }//if
+
+    /**
+     * Remove committed nodes from failed/prepared
+     */
+    cnoFailedNodes = clear_nodes(cnoFailedNodes, 
+				 cfailedNodes, 
+				 cnoCommitFailedNodes, 
+				 ccommitFailedNodes);
+    cnoPrepFailedNodes = clear_nodes(cnoPrepFailedNodes, 
+				     cprepFailedNodes,
+				     cnoCommitFailedNodes,
+				     ccommitFailedNodes);
     cnoCommitFailedNodes = 0;
   }//if
   /**-----------------------------------------------------------------------
@@ -4705,6 +4764,14 @@ Qmgr::execDUMP_STATE_ORD(Signal* signal)
   default:
     ;
   }//switch
+
+#ifdef ERROR_INSERT
+  if (signal->theData[0] == 935 && signal->getLength() == 2)
+  {
+    SET_ERROR_INSERT_VALUE(935);
+    c_error_insert_extra = signal->theData[1];
+  }
+#endif
 }//Qmgr::execDUMP_STATE_ORD()
 
 void Qmgr::execSET_VAR_REQ(Signal* signal) 

ndb/src/kernel/vm/Configuration.cpp

@@ -479,7 +479,7 @@ Configuration::setupConfiguration(){
   DBUG_VOID_RETURN;
 }
 
-bool 
+Uint32
 Configuration::lockPagesInMainMemory() const {
   return _lockPagesInMainMemory;
 }

ndb/src/kernel/vm/Configuration.hpp

@@ -36,7 +36,7 @@ public:
   void setupConfiguration();
   void closeConfiguration(bool end_session= true);
   
-  bool lockPagesInMainMemory() const;
+  Uint32 lockPagesInMainMemory() const;
   
   int timeBetweenWatchDogCheck() const ;
   void timeBetweenWatchDogCheck(int value);

ndb/src/mgmapi/mgmapi.cpp

@@ -93,6 +93,7 @@ struct ndb_mgm_handle {
   char last_error_desc[NDB_MGM_MAX_ERR_DESC_SIZE];
   int read_timeout;
   int write_timeout;
+  unsigned int connect_timeout;
 
   NDB_SOCKET_TYPE socket;
 
@@ -159,6 +160,7 @@ ndb_mgm_create_handle()
   h->socket          = NDB_INVALID_SOCKET;
   h->read_timeout    = 50000;
   h->write_timeout   = 100;
+  h->connect_timeout = 0;
   h->cfg_i           = -1;
   h->errstream       = stdout;
   h->m_name          = 0;
@@ -426,6 +428,16 @@ int ndb_mgm_is_connected(NdbMgmHandle handle)
   return handle->connected;
 }
 
+extern "C"
+int ndb_mgm_set_connect_timeout(NdbMgmHandle handle, unsigned int seconds)
+{
+  if(!handle)
+    return -1;
+
+  handle->connect_timeout= seconds;
+  return 0;
+}
+
 /**
  * Connect to a management server
  */
@@ -456,6 +468,7 @@ ndb_mgm_connect(NdbMgmHandle handle, int no_retries,
   Uint32 i;
   int binderror = 0;
   SocketClient s(0, 0);
+  s.set_connect_timeout(handle->connect_timeout);
   if (!s.init())
   {
     fprintf(handle->errstream, 

ndb/src/mgmsrv/ConfigInfo.cpp

@@ -563,10 +563,10 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     "If set to yes, then NDB Cluster data will not be swapped out to disk",
     ConfigInfo::CI_USED,
     true,
-    ConfigInfo::CI_BOOL,
-    "false",
-    "false",
-    "true" },
+    ConfigInfo::CI_INT,
+    "0",
+    "1",
+    "2" },
 
   {
     CFG_DB_WATCHDOG_INTERVAL,

ndb/src/mgmsrv/MgmtSrvr.cpp

@@ -137,8 +137,11 @@ MgmtSrvr::logLevelThreadRun()
         m_started_nodes.erase(0, false);
         m_started_nodes.unlock();
 
-        setEventReportingLevelImpl(node, req);
-
+        if (setEventReportingLevelImpl(node, req))
+	{
+	  ndbout_c("setEventReportingLevelImpl(%d): failed", node);
+	}
+	
         SetLogLevelOrd ord;
         ord = m_nodeLogLevel[node];
         setNodeLogLevelImpl(node, ord);
@@ -155,10 +158,16 @@ MgmtSrvr::logLevelThreadRun()
       m_log_level_requests.erase(0, false);
       m_log_level_requests.unlock();
 
-      if(req.blockRef == 0){
+      if(req.blockRef == 0)
+      {
 	req.blockRef = _ownReference;
-	setEventReportingLevelImpl(0, req);
-      } else {
+	if (setEventReportingLevelImpl(0, req))
+	{
+	  ndbout_c("setEventReportingLevelImpl: failed 2!");
+	}
+      } 
+      else 
+      {
         SetLogLevelOrd ord;
         ord = req;
 	setNodeLogLevelImpl(req.blockRef, ord);
@@ -1376,9 +1385,6 @@ int MgmtSrvr::restartDB(bool nostart, bool initialStart,
   NodeId nodeId = 0;
   NDB_TICKS maxTime = NdbTick_CurrentMillisecond() + waitTime;
 
-  ndbout_c(" %d", nodes.get(1));
-  ndbout_c(" %d", nodes.get(2));
-
   while(getNextNodeId(&nodeId, NDB_MGM_NODE_TYPE_NDB)) {
     if (!nodes.get(nodeId))
       continue;
@@ -1584,6 +1590,11 @@ MgmtSrvr::setEventReportingLevelImpl(int nodeId,
     }
   }
 
+  if (nodes.isclear())
+  {
+    return SEND_OR_RECEIVE_FAILED;
+  }
+  
   int error = 0;
   while (!nodes.isclear())
   {
@@ -1600,16 +1611,24 @@ MgmtSrvr::setEventReportingLevelImpl(int nodeId,
       error = 1;
       break;
     }
+      // Since sending okToSend(true), 
+      // there is no guarantee that NF_COMPLETEREP will come
+      // i.e listen also to NODE_FAILREP
+    case GSN_NODE_FAILREP: {
+      const NodeFailRep * const rep =
+	CAST_CONSTPTR(NodeFailRep, signal->getDataPtr());
+      NdbNodeBitmask mask;
+      mask.assign(NdbNodeBitmask::Size, rep->theNodes);
+      nodes.bitANDC(mask);
+      break;
+    }
+      
     case GSN_NF_COMPLETEREP:{
       const NFCompleteRep * const rep =
 	CAST_CONSTPTR(NFCompleteRep, signal->getDataPtr());
       nodes.clear(rep->failedNodeId);
       break;
     }
-    case GSN_NODE_FAILREP:{
-      // ignore, NF_COMPLETEREP will arrive later
-      break;
-    }
     default:
       report_unknown_signal(signal);
       return SEND_OR_RECEIVE_FAILED;
@@ -1909,7 +1928,10 @@ MgmtSrvr::handleStatus(NodeId nodeId, bool alive, bool nfComplete)
 
   theData[1] = nodeId;
   if (alive) {
-    m_started_nodes.push_back(nodeId);
+    if (nodeTypes[nodeId] == NODE_TYPE_DB)
+    {
+      m_started_nodes.push_back(nodeId);
+    }
     rep->setEventType(NDB_LE_Connected);
   } else {
     rep->setEventType(NDB_LE_Disconnected);

ndb/src/ndbapi/ClusterMgr.cpp

@@ -507,6 +507,7 @@ ClusterMgr::reportConnected(NodeId nodeId){
   theNode.m_info.m_version = 0;
   theNode.compatible = true;
   theNode.nfCompleteRep = true;
+  theNode.m_state.startLevel = NodeState::SL_NOTHING;
   
   theFacade.ReportNodeAlive(nodeId);
 }
@@ -518,14 +519,13 @@ ClusterMgr::reportDisconnected(NodeId nodeId){
 
   noOfConnectedNodes--;
   theNodes[nodeId].connected = false;
-
   theNodes[nodeId].m_state.m_connected_nodes.clear();
 
-  reportNodeFailed(nodeId);
+  reportNodeFailed(nodeId, true);
 }
 
 void
-ClusterMgr::reportNodeFailed(NodeId nodeId){
+ClusterMgr::reportNodeFailed(NodeId nodeId, bool disconnect){
 
   Node & theNode = theNodes[nodeId];
  
@@ -536,10 +536,11 @@ ClusterMgr::reportNodeFailed(NodeId nodeId){
   {
     theFacade.doDisconnect(nodeId);
   }
+  
   const bool report = (theNode.m_state.startLevel != NodeState::SL_NOTHING);  
   theNode.m_state.startLevel = NodeState::SL_NOTHING;
   
-  if(report)
+  if(disconnect || report)
   {
     theFacade.ReportNodeDead(nodeId);
   }

ndb/src/ndbapi/ClusterMgr.hpp

@@ -97,8 +97,8 @@ private:
   NdbMutex*     clusterMgrThreadMutex;
   
   void showState(NodeId nodeId);
-  void reportNodeFailed(NodeId nodeId);
-
+  void reportNodeFailed(NodeId nodeId, bool disconnect = false);
+  
   /**
    * Signals received
    */

ndb/src/ndbapi/SignalSender.cpp

@@ -19,6 +19,14 @@
 #include <signaldata/NFCompleteRep.hpp>
 #include <signaldata/NodeFailRep.hpp>
 
+static
+void
+require(bool x)
+{
+  if (!x)
+    abort();
+}
+
 SimpleSignal::SimpleSignal(bool dealloc){
   memset(this, 0, sizeof(* this));
   deallocSections = dealloc;
@@ -145,6 +153,7 @@ SignalSender::waitFor(Uint32 timeOutMillis, T & t)
 {
   SimpleSignal * s = t.check(m_jobBuffer);
   if(s != 0){
+    m_usedBuffer.push_back(s);
     return s;
   }
   

ndb/src/ndbapi/ndberror.c

@@ -208,7 +208,6 @@ ErrorBundle ErrorCodes[] = {
   /**
    * Internal errors
    */
-  { 892,  IE, "Inconsistent hash index. The index needs to be dropped and recreated" },
   { 896,  IE, "Tuple corrupted - wrong checksum or column data in invalid format" },
   { 901,  IE, "Inconsistent ordered index. The index needs to be dropped and recreated" },
   { 202,  IE, "202" },

ndb/test/ndbapi/testNodeRestart.cpp

@@ -954,6 +954,96 @@ int runBug24717(NDBT_Context* ctx, NDBT_Step* step){
   return NDBT_OK;
 }
 
+int runBug25364(NDBT_Context* ctx, NDBT_Step* step){
+  int result = NDBT_OK;
+  NdbRestarter restarter;
+  Ndb* pNdb = GETNDB(step);
+  int loops = ctx->getNumLoops();
+  
+  if (restarter.getNumDbNodes() < 4)
+    return NDBT_OK;
+
+  int val2[] = { DumpStateOrd::CmvmiSetRestartOnErrorInsert, 1 };
+
+  for (; loops; loops --)
+  {
+    int master = restarter.getMasterNodeId();
+    int victim = restarter.getRandomNodeOtherNodeGroup(master, rand());
+    int second = restarter.getRandomNodeSameNodeGroup(victim, rand());
+    
+    int dump[] = { 935, victim } ;
+    if (restarter.dumpStateOneNode(master, dump, 2))
+      return NDBT_FAILED;
+  
+    if (restarter.dumpStateOneNode(master, val2, 2))
+      return NDBT_FAILED;
+  
+    if (restarter.restartOneDbNode(second, false, true, true))
+      return NDBT_FAILED;
+
+    int nodes[2] = { master, second };
+    if (restarter.waitNodesNoStart(nodes, 2))
+      return NDBT_FAILED;
+
+    restarter.startNodes(nodes, 2);
+
+    if (restarter.waitNodesStarted(nodes, 2))
+      return NDBT_FAILED;
+  }
+  
+  return NDBT_OK;
+}
+
+int runBug25554(NDBT_Context* ctx, NDBT_Step* step){
+  
+  int result = NDBT_OK;
+  int loops = ctx->getNumLoops();
+  int records = ctx->getNumRecords();
+  NdbRestarter restarter;
+  
+  if (restarter.getNumDbNodes() < 4)
+    return NDBT_OK;
+
+  for (int i = 0; i<loops; i++)
+  {
+    int master = restarter.getMasterNodeId();
+    int node1 = restarter.getRandomNodeOtherNodeGroup(master, rand());
+    restarter.restartOneDbNode(node1, false, true, true);
+
+    int val2[] = { DumpStateOrd::CmvmiSetRestartOnErrorInsert, 1 };
+  
+    if (restarter.dumpStateOneNode(master, val2, 2))
+      return NDBT_FAILED;
+
+    if (restarter.insertErrorInNode(master, 7141))
+      return NDBT_FAILED;
+
+    if (restarter.waitNodesNoStart(&node1, 1))
+      return NDBT_FAILED;
+
+    if (restarter.dumpStateOneNode(node1, val2, 2))
+      return NDBT_FAILED;
+
+    if (restarter.insertErrorInNode(node1, 932))
+      return NDBT_FAILED;
+
+    if (restarter.startNodes(&node1, 1))
+      return NDBT_FAILED;
+
+    int nodes[] = { master, node1 };
+    if (restarter.waitNodesNoStart(nodes, 2))
+      return NDBT_FAILED;
+
+    if (restarter.startNodes(nodes, 2))
+      return NDBT_FAILED;
+
+    if (restarter.waitClusterStarted())
+      return NDBT_FAILED;
+  }    
+
+  return NDBT_OK;
+}
+
 
 NDBT_TESTSUITE(testNodeRestart);
 TESTCASE("NoLoad", 
@@ -1270,6 +1360,12 @@ TESTCASE("Bug20185",
 TESTCASE("Bug24717", ""){
   INITIALIZER(runBug24717);
 }
+TESTCASE("Bug25364", ""){
+  INITIALIZER(runBug25364);
+}
+TESTCASE("Bug25554", ""){
+  INITIALIZER(runBug25554);
+}
 NDBT_TESTSUITE_END(testNodeRestart);
 
 int main(int argc, const char** argv){

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

@@ -469,6 +469,14 @@ max-time: 1000
 cmd: testNodeRestart
 args: -n Bug24717 T1
 
+max-time: 1000
+cmd: testNodeRestart
+args: -n Bug25364 T1
+
+max-time: 1000
+cmd: testNodeRestart
+args: -n Bug25554 T1
+
 # OLD FLEX
 max-time: 500
 cmd: flexBench

sql/ha_ndbcluster.cc

@@ -1106,6 +1106,16 @@ int ha_ndbcluster::build_index_list(Ndb *ndb, TABLE *tab, enum ILBP phase)
 	error= create_unique_index(unique_index_name, key_info);
         break;
       case ORDERED_INDEX:
+	if (key_info->algorithm == HA_KEY_ALG_HASH)
+	{
+	  push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+			      ER_UNSUPPORTED_EXTENSION,
+			      ER(ER_UNSUPPORTED_EXTENSION),
+			      "Ndb does not support non-unique "
+			      "hash based indexes");
+	  error= HA_ERR_UNSUPPORTED;
+	  break;
+	}
         error= create_ordered_index(index_name, key_info);
         break;
       default:
@@ -2955,13 +2965,10 @@ KEY* key_info;
       DBUG_RETURN(error == HA_ERR_KEY_NOT_FOUND ? HA_ERR_END_OF_FILE : error);
     }
     else if (type == UNIQUE_INDEX)
-    {
-      error= unique_index_scan(key_info, 
-			       start_key->key, 
-			       start_key->length, 
-			       buf);
-      DBUG_RETURN(error == HA_ERR_KEY_NOT_FOUND ? HA_ERR_END_OF_FILE : error);
-    }
+      DBUG_RETURN(unique_index_scan(key_info, 
+				    start_key->key, 
+				    start_key->length, 
+				    buf));
     break;
   default:
     break;
@@ -3137,20 +3144,26 @@ void ha_ndbcluster::position(const byte *record)
       size_t len = key_part->length;
       const byte * ptr = record + key_part->offset;
       Field *field = key_part->field;
-      if ((field->type() ==  MYSQL_TYPE_VARCHAR) &&
-	  ((Field_varstring*)field)->length_bytes == 1)
+      if (field->type() ==  MYSQL_TYPE_VARCHAR)
       {
-	/** 
-	 * Keys always use 2 bytes length
-	 */
-	buff[0] = ptr[0];
-	buff[1] = 0;
-	memcpy(buff+2, ptr + 1, len);	
-	len += 2;
+        if (((Field_varstring*)field)->length_bytes == 1)
+        {
+          /**
+           * Keys always use 2 bytes length
+           */
+          buff[0] = ptr[0];
+          buff[1] = 0;
+          memcpy(buff+2, ptr + 1, len);
+        }
+        else
+        {
+          memcpy(buff, ptr, len + 2);
+        }
+        len += 2;
       }
       else
       {
-	memcpy(buff, ptr, len);
+        memcpy(buff, ptr, len);
       }
       buff += len;
     }
@@ -4172,19 +4185,29 @@ static int create_ndb_column(NDBCOL &col,
       col.setType(NDBCOL::Text);
       col.setCharset(cs);
     }
-    // Use "<=" even if "<" is the exact condition
-    if (field->max_length() <= (1 << 8))
-      goto mysql_type_tiny_blob;
-    else if (field->max_length() <= (1 << 16))
     {
-      col.setInlineSize(256);
-      col.setPartSize(2000);
-      col.setStripeSize(16);
+      Field_blob *field_blob= (Field_blob *)field;
+      /*
+       * max_data_length is 2^8-1, 2^16-1, 2^24-1 for tiny, blob, medium.
+       * Tinyblob gets no blob parts.  The other cases are just a crude
+       * way to control part size and striping.
+       *
+       * In mysql blob(256) is promoted to blob(65535) so it does not
+       * in fact fit "inline" in NDB.
+       */
+      if (field_blob->max_data_length() < (1 << 8))
+        goto mysql_type_tiny_blob;
+      else if (field_blob->max_data_length() < (1 << 16))
+      {
+        col.setInlineSize(256);
+        col.setPartSize(2000);
+        col.setStripeSize(16);
+      }
+      else if (field_blob->max_data_length() < (1 << 24))
+        goto mysql_type_medium_blob;
+      else
+        goto mysql_type_long_blob;
     }
-    else if (field->max_length() <= (1 << 24))
-      goto mysql_type_medium_blob;
-    else
-      goto mysql_type_long_blob;
     break;
   mysql_type_medium_blob:
   case MYSQL_TYPE_MEDIUM_BLOB:   

sql/mysqld.cc

@@ -4577,8 +4577,8 @@ enum options_mysqld
   OPT_LOG_BIN_TRUST_FUNCTION_CREATORS,
   OPT_SAFE_SHOW_DB, OPT_INNODB_SAFE_BINLOG,
   OPT_INNODB, OPT_ISAM,
-  OPT_ENGINE_CONDITION_PUSHDOWN,
-  OPT_NDBCLUSTER, OPT_NDB_CONNECTSTRING, OPT_NDB_USE_EXACT_COUNT,
+  OPT_ENGINE_CONDITION_PUSHDOWN, OPT_NDBCLUSTER, OPT_NDB_CONNECTSTRING, 
+  OPT_NDB_USE_EXACT_COUNT, OPT_NDB_USE_TRANSACTIONS,
   OPT_NDB_FORCE_SEND, OPT_NDB_AUTOINCREMENT_PREFETCH_SZ,
   OPT_NDB_SHM, OPT_NDB_OPTIMIZED_NODE_SELECTION, OPT_NDB_CACHE_CHECK_TIME,
   OPT_NDB_MGMD, OPT_NDB_NODEID,
@@ -5213,6 +5213,17 @@ Disable with --skip-ndbcluster (will save memory).",
    (gptr*) &global_system_variables.ndb_use_exact_count,
    (gptr*) &global_system_variables.ndb_use_exact_count,
    0, GET_BOOL, OPT_ARG, 1, 0, 0, 0, 0, 0},
+  {"ndb-use-transactions", OPT_NDB_USE_TRANSACTIONS,
+   "Use transactions for large inserts, if enabled then large "
+   "inserts will be split into several smaller transactions",
+   (gptr*) &global_system_variables.ndb_use_transactions,
+   (gptr*) &global_system_variables.ndb_use_transactions,
+   0, GET_BOOL, OPT_ARG, 1, 0, 0, 0, 0, 0},
+  {"ndb_use_transactions", OPT_NDB_USE_TRANSACTIONS,
+   "same as --ndb-use-transactions.",
+   (gptr*) &global_system_variables.ndb_use_transactions,
+   (gptr*) &global_system_variables.ndb_use_transactions,
+   0, GET_BOOL, OPT_ARG, 1, 0, 0, 0, 0, 0},
   {"ndb-shm", OPT_NDB_SHM,
    "Use shared memory connections when available.",
    (gptr*) &opt_ndb_shm,

sql/sql_class.cc

@@ -290,9 +290,6 @@ void THD::init(void)
 					       variables.date_format);
   variables.datetime_format= date_time_format_copy((THD*) 0,
 						   variables.datetime_format);
-#ifdef HAVE_NDBCLUSTER_DB
-  variables.ndb_use_transactions= 1;
-#endif
   pthread_mutex_unlock(&LOCK_global_system_variables);
   server_status= SERVER_STATUS_AUTOCOMMIT;
   if (variables.sql_mode & MODE_NO_BACKSLASH_ESCAPES)