Merge whalegate.ndb.mysql.com:/home/tomas/mysql-5.0

into  whalegate.ndb.mysql.com:/home/tomas/mysql-5.0-ndb-merge


mysql-test/r/view.result:
  Auto merged

mysql-test/r/ndb_alter_table2.result

@@ -40,3 +40,22 @@ a	b	c
 select * from t1;
 a	b	c
 drop table t1;
+DROP TABLE IF EXISTS truncate_test;
+CREATE TABLE truncate_test (
+i INT PRIMARY KEY,
+a INT,
+b VARCHAR(11),
+UNIQUE KEY (a) 
+) ENGINE = NDB;
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+TRUNCATE truncate_test;
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+SELECT * FROM truncate_test;
+i	a	b
+1	1	test
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+SELECT * FROM truncate_test;
+i	a	b
+1	1	new
+DROP TABLE truncate_test;

mysql-test/t/ndb_alter_table2.test

@@ -81,3 +81,34 @@ select * from t1;
 select * from t1;
 
 drop table t1;
+
+#For BUG#29851 TRUNCATE causes error 4350 from cluster in INSERT... ON DUPLICATE KEY UPDATE
+
+connection con1;
+
+--disable_warnings
+DROP TABLE IF EXISTS truncate_test;
+--enable_warnings
+
+CREATE TABLE truncate_test (
+  i INT PRIMARY KEY,
+  a INT,
+  b VARCHAR(11),
+  UNIQUE KEY (a) 
+) ENGINE = NDB;
+
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+
+connection con2;
+TRUNCATE truncate_test;
+
+connection con1;
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+SELECT * FROM truncate_test;
+
+connection con2;
+INSERT INTO truncate_test VALUES (1, 1, 'test') ON DUPLICATE KEY UPDATE b = 'new';
+SELECT * FROM truncate_test;
+
+DROP TABLE truncate_test;

ndb/include/kernel/ndb_limits.h

@@ -27,6 +27,17 @@
 #define MAX_NDB_NODES 49
 #define MAX_NODES     64
 
+/**************************************************************************
+ * IT SHOULD BE (MAX_NDB_NODES - 1).
+ * WHEN MAX_NDB_NODE IS CHANGED, IT SHOULD BE CHANGED ALSO
+ **************************************************************************/
+#define MAX_DATA_NODE_ID 48
+/**************************************************************************
+ * IT SHOULD BE (MAX_NODES - 1).
+ * WHEN MAX_NODES IS CHANGED, IT SHOULD BE CHANGED ALSO
+ **************************************************************************/
+#define MAX_NODES_ID 63
+
 /**
  * MAX_API_NODES = MAX_NODES - No of NDB Nodes in use
  */

ndb/include/mgmapi/ndb_logevent.h

@@ -551,7 +551,13 @@ extern "C" {
       /** Log event specific data for for corresponding NDB_LE_ log event */
       struct {
 	int      gth;
+        /* union is for compatibility backward.
+         * page_size_kb member variable should be removed in the future
+        */
+        union {
 	  unsigned page_size_kb;
+          unsigned page_size_bytes;
+        };
 	unsigned pages_used;
 	unsigned pages_total;
 	unsigned block;

ndb/include/util/Bitmask.hpp

@@ -126,6 +126,7 @@ public:
 
   /**
    * setField - Set bitfield at given position and length (max 32 bits)
+   * Note : length == 0 not supported.
    */
   static void setField(unsigned size, Uint32 data[],
       unsigned pos, unsigned len, Uint32 val);
@@ -133,6 +134,7 @@ public:
 
   /**
    * getField - Get bitfield at given position and length
+   * Note : length == 0 not supported.
    */
   static void getField(unsigned size, const Uint32 data[],
 		       unsigned pos, unsigned len, Uint32 dst[]);
@@ -814,6 +816,9 @@ BitmaskImpl::getField(unsigned size, const Uint32 src[],
 		      unsigned pos, unsigned len, Uint32 dst[])
 {
   assert(pos + len <= (size << 5));
+  assert (len != 0);
+  if (len == 0)
+    return;
 
   src += (pos >> 5);
   Uint32 offset = pos & 31;
@@ -833,6 +838,9 @@ BitmaskImpl::setField(unsigned size, Uint32 dst[],
 		      unsigned pos, unsigned len, const Uint32 src[])
 {
   assert(pos + len <= (size << 5));
+  assert(len != 0);
+  if (len == 0)
+    return;
 
   dst += (pos >> 5);
   Uint32 offset = pos & 31;

ndb/src/common/transporter/TCP_Transporter.cpp

@@ -337,22 +337,32 @@ TCP_Transporter::doSend() {
 
   // Empty the SendBuffers
   
+  bool sent_any = true;
+  while (m_sendBuffer.dataSize > 0)
+  {
     const char * const sendPtr = m_sendBuffer.sendPtr;
     const Uint32 sizeToSend    = m_sendBuffer.sendDataSize;
-  if (sizeToSend > 0){
     const int nBytesSent = inet_send(theSocket, sendPtr, sizeToSend, 0);
     
-    if (nBytesSent > 0) {
+    if (nBytesSent > 0) 
+    {
+      sent_any = true;
       m_sendBuffer.bytesSent(nBytesSent);
       
       sendCount ++;
       sendSize  += nBytesSent;
-      if(sendCount == reportFreq){
+      if(sendCount == reportFreq)
+      {
 	reportSendLen(get_callback_obj(), remoteNodeId, sendCount, sendSize);
 	sendCount = 0;
 	sendSize  = 0;
       }
-    } else {
+    } 
+    else 
+    {
+      if (nBytesSent < 0 && InetErrno == EAGAIN && sent_any)
+        break;
+
       // Send failed
 #if defined DEBUG_TRANSPORTER
       ndbout_c("Send Failure(disconnect==%d) to node = %d nBytesSent = %d "

ndb/src/common/util/Bitmask.cpp

@@ -16,19 +16,42 @@
 #include <Bitmask.hpp>
 #include <NdbOut.hpp>
 
-#ifndef __TEST_BITMASK__
-
 void
 BitmaskImpl::getFieldImpl(const Uint32 src[],
 			  unsigned shiftL, unsigned len, Uint32 dst[])
 {
+  /* Copy whole words of src to dst, shifting src left
+   * by shiftL.  Undefined bits of the last written dst word
+   * should be zeroed.
+   */
   assert(shiftL < 32);
 
   unsigned shiftR = 32 - shiftL;
   unsigned undefined = shiftL ? ~0 : 0;
 
+  /* Merge first word with previously set bits if there's a shift */
   * dst = shiftL ? * dst : 0;
 
+  /* Treat the zero-shift case separately to avoid
+   * trampling or reading past the end of src
+   */
+  if (shiftL == 0)
+  {
+    while(len >= 32)
+    {
+      * dst++ = * src++;
+      len -=32;
+    }
+
+    if (len != 0)
+    {
+      /* Last word has some bits set */
+      Uint32 mask= ((1 << len) -1); // 0000111
+      * dst = (* src) & mask;
+    }
+  }
+  else // shiftL !=0, need to build each word from two words shifted
+  {
     while(len >= 32)
     {
       * dst++ |= (* src) << shiftL;
@@ -36,15 +59,21 @@ BitmaskImpl::getFieldImpl(const Uint32 src[],
       len -= 32;
     }
 
-  if(len < shiftR)
+    /* Have space for shiftR more bits in the current dst word
+     * is that enough?
+     */
+    if(len <= shiftR)
     {
+      /* Fit the remaining bits in the current dst word */
       * dst |= ((* src) & ((1 << len) - 1)) << shiftL;
     }
     else
     {
+      /* Need to write to two dst words */
       * dst++ |= ((* src) << shiftL);
       * dst = ((* src) >> shiftR) & ((1 << (len - shiftR)) - 1) & undefined;
     }
+  }
 }
 
 void
@@ -66,301 +95,23 @@ BitmaskImpl::setFieldImpl(Uint32 dst[],
     len -= 32;
   }
   
+  /* Copy last bits */
   Uint32 mask = ((1 << len) -1);
   * dst = (* dst & ~mask);
-  if(len < shiftR)
+  if(len <= shiftR)
   {
+    /* Remaining bits fit in current word */
     * dst |= ((* src++) >> shiftL) & mask;
   }
   else
   {
+    /* Remaining bits update 2 words */
     * dst |= ((* src++) >> shiftL);
     * dst |= ((* src) & ((1 << (len - shiftR)) - 1)) << shiftR ;
   }
 }
-#else
-
-static
-void print(const Uint32 src[], Uint32 len, Uint32 pos = 0)
-{
-  printf("b'");
-  for(unsigned i = 0; i<len; i++)
-  {
-    if(BitmaskImpl::get((pos + len + 31) >> 5, src, i+pos))
-      printf("1");
-    else
-      printf("0");
-    if((i & 31) == 31)
-      printf(" ");
-  }
-}
-
-#define DEBUG 0
-#include <Vector.hpp>
-static void do_test(int bitmask_size);
-
-int
-main(int argc, char** argv)
-{
-  int loops = argc > 1 ? atoi(argv[1]) : 1000;
-  int max_size = argc > 2 ? atoi(argv[2]) : 1000;
-
-  
-  for(int i = 0; i<loops; i++)
-    do_test(1 + (rand() % max_size));
-}
-
-struct Alloc
-{
-  Uint32 pos;
-  Uint32 size;
-  Vector<Uint32> data;
-};
-
-static void require(bool b)
-{
-  if(!b) abort();
-}
-
-static
-bool cmp(const Uint32 b1[], const Uint32 b2[], Uint32 len)
-{
-  Uint32 sz32 = (len + 31) >> 5;
-  for(int i = 0; i<len; i++)
-  {
-    if(BitmaskImpl::get(sz32, b1, i) ^ BitmaskImpl::get(sz32, b2, i))
-      return false;
-  }
-  return true;
-}
-
-
-static int val_pos = 0;
-static int val[] = { 384, 241, 32, 
-		     1,1,1,1, 0,0,0,0, 1,1,1,1, 0,0,0,0,
-		     241 };
-
-static int lrand()
-{
-#if 0
-  return val[val_pos++];
-#else
-  return rand();
-#endif
-}
-
-static
-void rand(Uint32 dst[], Uint32 len)
-{
-  for(int i = 0; i<len; i++)
-    BitmaskImpl::set((len + 31) >> 5, dst, i, (lrand() % 1000) > 500);
-}
-
-static
-void simple(int pos, int size)
-{
-  ndbout_c("simple pos: %d size: %d", pos, size);
-  Vector<Uint32> _mask;
-  Vector<Uint32> _src;
-  Vector<Uint32> _dst;
-  Uint32 sz32 = (size + pos + 32) >> 5;
-  const Uint32 sz = 4 * sz32;
-  
-  Uint32 zero = 0;
-  _mask.fill(sz32+1, zero);
-  _src.fill(sz32+1, zero);
-  _dst.fill(sz32+1, zero);
-
-  Uint32 * src = _src.getBase();
-  Uint32 * dst = _dst.getBase();
-  Uint32 * mask = _mask.getBase();
-
-  memset(src, 0x0, sz);
-  memset(dst, 0x0, sz);
-  memset(mask, 0xFF, sz);
-  rand(src, size);
-  BitmaskImpl::setField(sz32, mask, pos, size, src);
-  BitmaskImpl::getField(sz32, mask, pos, size, dst);
-  printf("src: "); print(src, size+31); printf("\n");
-  printf("msk: "); print(mask, (sz32 << 5) + 31); printf("\n");
-  printf("dst: "); print(dst, size+31); printf("\n");
-  require(cmp(src, dst, size+31));
-};
 
-static
-void simple2(int size, int loops)
-{
-  ndbout_c("simple2 %d - ", size);
-  Vector<Uint32> _mask;
-  Vector<Uint32> _src;
-  Vector<Uint32> _dst;
-
-  Uint32 sz32 = (size + 32) >> 5;
-  Uint32 sz = sz32 << 2;
-  
-  Uint32 zero = 0;
-  _mask.fill(sz32+1, zero);
-  _src.fill(sz32+1, zero);
-  _dst.fill(sz32+1, zero);
-
-  Uint32 * src = _src.getBase();
-  Uint32 * dst = _dst.getBase();
-  Uint32 * mask = _mask.getBase();
-
-  Vector<Uint32> save;
-  for(int i = 0; i<loops; i++)
-  {
-    memset(mask, 0xFF, sz);
-    memset(dst, 0xFF, sz);
-    int len;
-    int pos = 0;
-    while(pos+1 < size)
-    {
-      memset(src, 0xFF, sz);
-      while(!(len = rand() % (size - pos)));
-      BitmaskImpl::setField(sz32, mask, pos, len, src);
-      if(memcmp(dst, mask, sz))
-      {
-	ndbout_c("pos: %d len: %d", pos, len);
-	print(mask, size);
-	abort();
-      }
-      printf("[ %d %d ]", pos, len);
-      save.push_back(pos);
-      save.push_back(len);
-      pos += len;
-    }
-
-    for(int j = 0; j<save.size(); )
-    {
-      pos = save[j++];
-      len = save[j++];
-      memset(src, 0xFF, sz);
-      BitmaskImpl::getField(sz32, mask, pos, len, src);
-      if(memcmp(dst, src, sz))
-      {
-	ndbout_c("pos: %d len: %d", pos, len);
-	printf("src: "); print(src, size); printf("\n");
-	printf("dst: "); print(dst, size); printf("\n");
-	printf("msk: "); print(mask, size); printf("\n");
-	abort();
-      }
-    }
-    ndbout_c("");
-  }
-}
-
-static void 
-do_test(int bitmask_size)
-{
-#if 1
-  simple(rand() % 33, (rand() % 63)+1);
-//#else
-  Vector<Alloc> alloc_list;
-  bitmask_size = (bitmask_size + 31) & ~31;
-  Uint32 sz32 = (bitmask_size >> 5);
-  Vector<Uint32> alloc_mask;
-  Vector<Uint32> test_mask;
-  
-  ndbout_c("Testing bitmask of size %d", bitmask_size);
-  Uint32 zero = 0;
-  alloc_mask.fill(sz32, zero);
-  test_mask.fill(sz32, zero);
-  
-  for(int i = 0; i<5000; i++)
-  {
-    Vector<Uint32> tmp;
-    tmp.fill(sz32, zero);
-
-    int pos = lrand() % (bitmask_size - 1);
-    int free = 0;
-    if(BitmaskImpl::get(sz32, alloc_mask.getBase(), pos))
-    {
-      // Bit was allocated
-      // 1) Look up allocation
-      // 2) Check data
-      // 3) free it
-      size_t j;
-      int min, max;
-      for(j = 0; j<alloc_list.size(); j++)
-      {
-	min = alloc_list[j].pos;
-	max = min + alloc_list[j].size;
-	if(pos >= min && pos < max)
-	{
-	  break;
-	}
-      }
-      require(pos >= min && pos < max);
-      BitmaskImpl::getField(sz32, test_mask.getBase(), min, max-min, 
-			    tmp.getBase());
-      if(DEBUG)
-      {
-	printf("freeing [ %d %d ]", min, max);
-	printf("- mask: ");
-	print(tmp.getBase(), max - min);
-	
-	printf(" save: ");
-	size_t k;
-	Alloc& a = alloc_list[j];
-	for(k = 0; k<a.data.size(); k++)
-	  printf("%.8x ", a.data[k]);
-	printf("\n");
-      }
-      int bytes = (max - min + 7) >> 3;
-      if(!cmp(tmp.getBase(), alloc_list[j].data.getBase(), max - min))
-      {
-	abort();
-      }
-      while(min < max)
-	BitmaskImpl::clear(sz32, alloc_mask.getBase(), min++);
-      alloc_list.erase(j);
-    }
-    else
-    {
-      Vector<Uint32> tmp;
-      tmp.fill(sz32, zero);
-      
-      // Bit was free
-      // 1) Check how much space is avaiable
-      // 2) Create new allocation of lrandom size
-      // 3) Fill data with lrandom data
-      // 4) Update alloc mask
-      while(pos+free < bitmask_size && 
-	    !BitmaskImpl::get(sz32, alloc_mask.getBase(), pos+free))
-	free++;
-
-      Uint32 sz = 
-	(free <= 64 && ((lrand() % 100) > 80)) ? free : (lrand() % free); 
-      sz = sz ? sz : 1;
-      sz = pos + sz == bitmask_size ? sz - 1 : sz;
-      Alloc a;
-      a.pos = pos;
-      a.size = sz;
-      a.data.fill(((sz+31)>> 5)-1, zero);
-      if(DEBUG)
-	printf("pos %d -> alloc [ %d %d ]", pos, pos, pos+sz);
-      for(size_t j = 0; j<sz; j++)
-      {
-	BitmaskImpl::set(sz32, alloc_mask.getBase(), pos+j);
-	if((lrand() % 1000) > 500)
-	  BitmaskImpl::set((sz + 31) >> 5, a.data.getBase(), j);
-      }
-      if(DEBUG)
-      {
-	printf("- mask: ");
-	print(a.data.getBase(), sz);
-	printf("\n");
-      }
-      BitmaskImpl::setField(sz32, test_mask.getBase(), pos, sz, 
-			    a.data.getBase());
-      alloc_list.push_back(a);
-    }
-  }
-#endif
-}
-
-template class Vector<Alloc>;
-template class Vector<Uint32>;
-
-#endif
+/* Bitmask testcase code moved from here to
+ * storage/ndb/test/ndbapi/testBitfield.cpp
+ * to get coverage from automated testing
+ */

ndb/src/mgmapi/ndb_logevent.cpp

@@ -256,7 +256,7 @@ struct Ndb_logevent_body_row ndb_logevent_body[]= {
   ROW( ReceiveBytesStatistic, "mean_received_bytes", 2, mean_received_bytes),
 
   ROW( MemoryUsage, "gth",          1, gth),
-  ROW( MemoryUsage, "page_size_kb", 2, page_size_kb),
+  ROW( MemoryUsage, "page_size_bytes", 2, page_size_bytes),
   ROW( MemoryUsage, "pages_used",   3, pages_used),
   ROW( MemoryUsage, "pages_total",  4, pages_total),
   ROW( MemoryUsage, "block",        5, block),

ndb/src/mgmclient/CommandInterpreter.cpp

@@ -921,10 +921,14 @@ event_thread_run(void* p)
   {
     do_event_thread= 1;
     do {
-      if (ndb_logevent_get_next(log_handle, &log_event, 2000) <= 0)
-        continue;
+      int res= ndb_logevent_get_next(log_handle, &log_event, 2000);
+      if (res > 0)
+      {
         Guard g(printmutex);
         printLogEvent(&log_event);
+      }
+      else if (res < 0)
+        break;
     } while(do_event_thread);
     ndb_mgm_destroy_logevent_handle(&log_handle);
   }
@@ -2722,8 +2726,9 @@ CommandInterpreter::executeStartBackup(char* parameters, bool interactive)
   {
     int count = 0;
     int retry = 0;
+    int res;
     do {
-      if (ndb_logevent_get_next(log_handle, &log_event, 60000) > 0)
+      if ((res= ndb_logevent_get_next(log_handle, &log_event, 60000)) > 0)
       {
         int print = 0;
         switch (log_event.type) {
@@ -2753,7 +2758,7 @@ CommandInterpreter::executeStartBackup(char* parameters, bool interactive)
       {
         retry++;
       }
-    } while(count < 2 && retry < 3);
+    } while(res >= 0 && count < 2 && retry < 3);
 
     if (retry >= 3)
       ndbout << "get backup event failed for " << retry << " times" << endl;

ndb/src/mgmsrv/ConfigInfo.cpp

@@ -398,7 +398,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     ConfigInfo::CI_INT,
     MANDATORY,
     "1",
-    STR_VALUE(MAX_NODES) },
+    STR_VALUE(MAX_DATA_NODE_ID) },
 
   {
     CFG_NODE_ID,
@@ -410,7 +410,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     ConfigInfo::CI_INT,
     MANDATORY,
     "1",
-    STR_VALUE(MAX_NODES) },
+    STR_VALUE(MAX_DATA_NODE_ID) },
 
   {
     KEY_INTERNAL,
@@ -1261,7 +1261,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     ConfigInfo::CI_INT,
     MANDATORY,
     "1",
-    STR_VALUE(MAX_NODES) },
+    STR_VALUE(MAX_NODES_ID) },
 
   {
     CFG_NODE_ID,
@@ -1273,7 +1273,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     ConfigInfo::CI_INT,
     MANDATORY,
     "1",
-    STR_VALUE(MAX_NODES) },
+    STR_VALUE(MAX_NODES_ID) },
 
   {
     KEY_INTERNAL,
@@ -1404,7 +1404,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     ConfigInfo::CI_INT,
     MANDATORY,
     "1",
-    STR_VALUE(MAX_NODES) },
+    STR_VALUE(MAX_NODES_ID) },
   
   {
     CFG_NODE_ID,
@@ -1416,7 +1416,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     ConfigInfo::CI_INT,
     MANDATORY,
     "1",
-    STR_VALUE(MAX_NODES) },
+    STR_VALUE(MAX_NODES_ID) },
   
   {
     CFG_LOG_DESTINATION,

ndb/test/ndbapi/testBitfield.cpp

@@ -4,6 +4,8 @@
 #include <NDBT.hpp>
 #include <NdbApi.hpp>
 #include <HugoTransactions.hpp>
+#include <Bitmask.hpp>
+#include <Vector.hpp>
 
 static const char* _dbname = "TEST_DB";
 static int g_loops = 7;
@@ -37,6 +39,7 @@ static int unique_indexes(Ndb*, const NdbDictionary::Table* tab);
 static int ordered_indexes(Ndb*, const NdbDictionary::Table* tab);
 static int node_restart(Ndb*, const NdbDictionary::Table* tab);
 static int system_restart(Ndb*, const NdbDictionary::Table* tab);
+static int testBitmask();
 
 int 
 main(int argc, char** argv){
@@ -49,6 +52,15 @@ main(int argc, char** argv){
 			       ndb_std_get_one_option)))
     return NDBT_ProgramExit(NDBT_WRONGARGS);
 
+  int res = NDBT_FAILED;
+
+  /* Run cluster-independent tests */
+  for (int i=0; i<(10*g_loops); i++)
+  {
+    if (NDBT_OK != (res= testBitmask()))
+      return NDBT_ProgramExit(res);
+  }
+  
   Ndb_cluster_connection con(opt_connect_str);
   if(con.connect(12, 5, 1))
   {
@@ -60,7 +72,6 @@ main(int argc, char** argv){
   pNdb = new Ndb(&con, _dbname);  
   pNdb->init();
   while (pNdb->waitUntilReady() != 0);
-  int res = NDBT_FAILED;
 
   NdbDictionary::Dictionary * dict = pNdb->getDictionary();
 
@@ -121,14 +132,12 @@ create_random_table(Ndb* pNdb)
   do {
     NdbDictionary::Table tab;
     Uint32 cols = 1 + (rand() % (NDB_MAX_ATTRIBUTES_IN_TABLE - 1));
-    Uint32 keys = NDB_MAX_NO_OF_ATTRIBUTES_IN_KEY;
     Uint32 length = 4090;
-    Uint32 key_size = NDB_MAX_KEYSIZE_IN_WORDS;
     
     BaseString name; 
     name.assfmt("TAB_%d", rand() & 65535);
     tab.setName(name.c_str());
-    for(int i = 0; i<cols && length > 2; i++)
+    for(Uint32 i = 0; i<cols && length > 2; i++)
     {
       NdbDictionary::Column col;
       name.assfmt("COL_%d", i);
@@ -206,3 +215,393 @@ system_restart(Ndb* pNdb, const NdbDictionary::Table* tab)
 {
   return 0;
 }
+
+/* Note : folowing classes test functionality of storage/ndb/src/common/util/Bitmask.cpp
+ * and were originally defined there.
+ * Set BITMASK_DEBUG to 1 to get more test debugging info.
+ */
+#define BITMASK_DEBUG 0
+
+static
+bool cmp(const Uint32 b1[], const Uint32 b2[], Uint32 len)
+{
+  Uint32 sz32 = (len + 31) >> 5;
+  for(Uint32 i = 0; i<len; i++)
+  {
+    if(BitmaskImpl::get(sz32, b1, i) ^ BitmaskImpl::get(sz32, b2, i))
+      return false;
+  }
+  return true;
+}
+
+static
+void print(const Uint32 src[], Uint32 len, Uint32 pos = 0)
+{
+  printf("b'");
+  for(unsigned i = 0; i<len; i++)
+  {
+    if(BitmaskImpl::get((pos + len + 31) >> 5, src, i+pos))
+      printf("1");
+    else
+      printf("0");
+    if((i & 31) == 31)
+      printf(" ");
+  }
+}
+
+static int lrand()
+{
+  return rand();
+}
+
+static
+void rand(Uint32 dst[], Uint32 len)
+{
+  for(Uint32 i = 0; i<len; i++)
+    BitmaskImpl::set((len + 31) >> 5, dst, i, (lrand() % 1000) > 500);
+}
+
+static
+int checkNoTramplingGetSetField(const Uint32 totalTests)
+{
+  const Uint32 numWords= 67;
+  const Uint32 maxBitsToCopy= (numWords * 32);
+  Uint32 sourceBuf[numWords];
+  Uint32 targetBuf[numWords];
+
+  ndbout << "Testing : Bitmask NoTrampling\n";
+
+  memset(sourceBuf, 0x00, (numWords*4));
+
+  for (Uint32 test=0; test<totalTests; test++)
+  {
+    /* Always copy at least 1 bit */
+    Uint32 srcStart= rand() % (maxBitsToCopy -1);
+    Uint32 length= (rand() % ((maxBitsToCopy -1) - srcStart)) + 1;
+
+    if (BITMASK_DEBUG)
+      ndbout << "Testing start %u, length %u \n"
+             << srcStart
+             << length;
+    // Set target to all ones.
+    memset(targetBuf, 0xff, (numWords*4));
+
+    BitmaskImpl::getField(numWords, sourceBuf, srcStart, length, targetBuf);
+
+    // Check that there is no trampling
+    Uint32 firstUntrampledWord= (length + 31)/32;
+
+    for (Uint32 word=0; word< numWords; word++)
+    {
+      Uint32 targetWord= targetBuf[word];
+      if (BITMASK_DEBUG)
+        ndbout << "word=%d, targetWord=%u, firstUntrampledWord..=%u"
+               << word << targetWord << firstUntrampledWord;
+
+      if (! (word < firstUntrampledWord) ?
+          (targetWord == 0) :
+          (targetWord == 0xffffffff))
+      {
+        ndbout << "Notrampling getField failed for srcStart "
+               << srcStart
+               << " length " << length
+               << " at word " << word << "\n";
+        ndbout << "word=%d, targetWord=%u, firstUntrampledWord..=%u"
+               << word << targetWord << firstUntrampledWord;
+        return -1;
+      }
+
+    }
+
+    /* Set target back to all ones. */
+    memset(targetBuf, 0xff, (numWords*4));
+
+    BitmaskImpl::setField(numWords, targetBuf, srcStart, length, sourceBuf);
+
+    /* Check we've got all ones, with zeros only where expected */
+    for (Uint32 word=0; word< numWords; word++)
+    {
+      Uint32 targetWord= targetBuf[word];
+
+      for (Uint32 bit=0; bit< 32; bit++)
+      {
+        Uint32 bitNum= (word << 5) + bit;
+        bool expectedValue= !((bitNum >= srcStart) &&
+                              (bitNum < (srcStart + length)));
+        bool actualValue= (((targetWord >> bit) & 1) == 1);
+        if (BITMASK_DEBUG)
+          ndbout << "bitNum=%u expectedValue=%u, actual value=%u"
+                 << bitNum << expectedValue << actualValue;
+
+        if (actualValue != expectedValue)
+        {
+          ndbout << "Notrampling setField failed for srcStart "
+                 << srcStart
+                 << " length " << length
+                 << " at word " << word << " bit " << bit <<  "\n";
+          ndbout << "bitNum=%u expectedValue=%u, actual value=%u"
+                 << bitNum << expectedValue << actualValue;
+          return -1;
+        }
+      }
+    }
+
+  }
+
+  return 0;
+}
+
+static
+int simple(int pos, int size)
+{
+  ndbout << "Testing : Bitmask simple pos: " << pos << " size: " << size << "\n";
+  Vector<Uint32> _mask;
+  Vector<Uint32> _src;
+  Vector<Uint32> _dst;
+  Uint32 sz32 = (size + pos + 32) >> 5;
+  const Uint32 sz = 4 * sz32;
+
+  Uint32 zero = 0;
+  _mask.fill(sz32+1, zero);
+  _src.fill(sz32+1, zero);
+  _dst.fill(sz32+1, zero);
+
+  Uint32 * src = _src.getBase();
+  Uint32 * dst = _dst.getBase();
+  Uint32 * mask = _mask.getBase();
+
+  memset(src, 0x0, sz);
+  memset(dst, 0x0, sz);
+  memset(mask, 0xFF, sz);
+  rand(src, size);
+  BitmaskImpl::setField(sz32, mask, pos, size, src);
+  BitmaskImpl::getField(sz32, mask, pos, size, dst);
+  if (BITMASK_DEBUG)
+  {
+    printf("src: "); print(src, size+31); printf("\n");
+    printf("msk: "); print(mask, (sz32 << 5) + 31); printf("\n");
+    printf("dst: "); print(dst, size+31); printf("\n");
+  }
+  return (cmp(src, dst, size+31)?0 : -1);
+};
+
+struct Alloc
+{
+  Uint32 pos;
+  Uint32 size;
+  Vector<Uint32> data;
+};
+
+static
+int
+testRanges(Uint32 bitmask_size)
+{
+  Vector<Alloc> alloc_list;
+  bitmask_size = (bitmask_size + 31) & ~31;
+  Uint32 sz32 = (bitmask_size >> 5);
+  Vector<Uint32> alloc_mask;
+  Vector<Uint32> test_mask;
+
+  ndbout_c("Testing : Bitmask ranges for bitmask of size %d", bitmask_size);
+  Uint32 zero = 0;
+  alloc_mask.fill(sz32, zero);
+  test_mask.fill(sz32, zero);
+
+  /* Loop a number of times, setting and clearing bits in the mask
+   * and tracking the modifications in a separate structure.
+   * Check that both structures remain in sync
+   */
+  for(int i = 0; i<5000; i++)
+  {
+    Vector<Uint32> tmp;
+    tmp.fill(sz32, zero);
+
+    Uint32 pos = lrand() % (bitmask_size - 1);
+    Uint32 free = 0;
+    if(BitmaskImpl::get(sz32, alloc_mask.getBase(), pos))
+    {
+      // Bit was allocated
+      // 1) Look up allocation
+      // 2) Check data
+      // 3) free it
+      size_t j;
+      Uint32 min, max;
+      for(j = 0; j<alloc_list.size(); j++)
+      {
+	min = alloc_list[j].pos;
+	max = min + alloc_list[j].size;
+	if(pos >= min && pos < max)
+	{
+	  break;
+	}
+      }
+      if (! ((pos >= min) && (pos < max)))
+      {
+        printf("Failed with pos %u, min %u, max %u\n",
+               pos, min, max);
+        return -1;
+      }
+      BitmaskImpl::getField(sz32, test_mask.getBase(), min, max-min,
+			    tmp.getBase());
+      if(BITMASK_DEBUG)
+      {
+	printf("freeing [ %d %d ]", min, max);
+	printf("- mask: ");
+	print(tmp.getBase(), max - min);
+
+	printf(" save: ");
+	size_t k;
+	Alloc& a = alloc_list[j];
+	for(k = 0; k<a.data.size(); k++)
+	  printf("%.8x ", a.data[k]);
+	printf("\n");
+      }
+      if(!cmp(tmp.getBase(), alloc_list[j].data.getBase(), max - min))
+      {
+	return -1;
+      }
+      while(min < max)
+	BitmaskImpl::clear(sz32, alloc_mask.getBase(), min++);
+      alloc_list.erase(j);
+    }
+    else
+    {
+      Vector<Uint32> tmp;
+      tmp.fill(sz32, zero);
+
+      // Bit was free
+      // 1) Check how much space is avaiable
+      // 2) Create new allocation of lrandom size
+      // 3) Fill data with lrandom data
+      // 4) Update alloc mask
+      while(pos+free < bitmask_size &&
+	    !BitmaskImpl::get(sz32, alloc_mask.getBase(), pos+free))
+	free++;
+
+      Uint32 sz =
+	(free <= 64 && ((lrand() % 100) > 80)) ? free : (lrand() % free);
+      sz = sz ? sz : 1;
+      sz = pos + sz == bitmask_size ? sz - 1 : sz;
+      Alloc a;
+      a.pos = pos;
+      a.size = sz;
+      a.data.fill(((sz+31)>> 5)-1, zero);
+      if(BITMASK_DEBUG)
+	printf("pos %d -> alloc [ %d %d ]", pos, pos, pos+sz);
+      for(size_t j = 0; j<sz; j++)
+      {
+	BitmaskImpl::set(sz32, alloc_mask.getBase(), pos+j);
+	if((lrand() % 1000) > 500)
+	  BitmaskImpl::set((sz + 31) >> 5, a.data.getBase(), j);
+      }
+      if(BITMASK_DEBUG)
+      {
+	printf("- mask: ");
+	print(a.data.getBase(), sz);
+	printf("\n");
+      }
+      BitmaskImpl::setField(sz32, test_mask.getBase(), pos, sz,
+			    a.data.getBase());
+      alloc_list.push_back(a);
+    }
+  }
+
+#ifdef NDB_BM_SUPPORT_RANGE
+  for(Uint32 i = 0; i<1000; i++)
+  {
+    Uint32 sz32 = 10+rand() % 100;
+    Uint32 zero = 0;
+    Vector<Uint32> map;
+    map.fill(sz32, zero);
+
+    Uint32 sz = 32 * sz32;
+    Uint32 start = (rand() % sz);
+    Uint32 stop = start + ((rand() % (sz - start)) & 0xFFFFFFFF);
+
+    Vector<Uint32> check;
+    check.fill(sz32, zero);
+
+    /* Verify range setting method works correctly */
+    for(Uint32 j = 0; j<sz; j++)
+    {
+      bool expect = (j >= start && j<stop);
+      if(expect)
+	BitmaskImpl::set(sz32, check.getBase(), j);
+    }
+
+    BitmaskImpl::set_range(sz32, map.getBase(), start, stop);
+    if (!BitmaskImpl::equal(sz32, map.getBase(), check.getBase()))
+    {
+      ndbout_c(" FAIL 1 sz: %d [ %d %d ]", sz, start, stop);
+      printf("check: ");
+      for(Uint32 j = 0; j<sz32; j++)
+	printf("%.8x ", check[j]);
+      printf("\n");
+
+      printf("map  : ");
+      for(Uint32 j = 0; j<sz32; j++)
+	printf("%.8x ", map[j]);
+      printf("\n");
+      return -1;
+    }
+
+    map.clear();
+    check.clear();
+
+    /* Verify range clearing method works correctly */
+    Uint32 one = ~(Uint32)0;
+    map.fill(sz32, one);
+    check.fill(sz32, one);
+
+    for(Uint32 j = 0; j<sz; j++)
+    {
+      bool expect = (j >= start && j<stop);
+      if(expect)
+	BitmaskImpl::clear(sz32, check.getBase(), j);
+    }
+
+    BitmaskImpl::clear_range(sz32, map.getBase(), start, stop);
+    if (!BitmaskImpl::equal(sz32, map.getBase(), check.getBase()))
+    {
+      ndbout_c(" FAIL 2 sz: %d [ %d %d ]", sz, start, stop);
+      printf("check: ");
+      for(Uint32 j = 0; j<sz32; j++)
+	printf("%.8x ", check[j]);
+      printf("\n");
+
+      printf("map  : ");
+      for(Uint32 j = 0; j<sz32; j++)
+	printf("%.8x ", map[j]);
+      printf("\n");
+      return -1;
+    }
+  }
+#endif
+
+  return 0;
+}
+
+static
+int
+testBitmask()
+{
+  /* Some testcases from storage/ndb/src/common/util/Bitmask.cpp */
+  int res= 0;
+
+  if ((res= checkNoTramplingGetSetField(100 /* totalTests */)) != 0)
+    return res;
+
+  if ((res= simple(rand() % 33, // position
+                   (rand() % 63)+1) // size
+       ) != 0)
+    return res;
+
+  if ((res= testRanges(1+(rand() % 1000) // bitmask size
+                       )) != 0)
+    return res;
+
+  return 0;
+}
+
+template class Vector<Alloc>;
+template class Vector<Uint32>;

sql/ha_ndbcluster.cc

@@ -439,7 +439,8 @@ void ha_ndbcluster::no_uncommitted_rows_reset(THD *thd)
 
 void ha_ndbcluster::invalidate_dictionary_cache(bool global)
 {
-  NDBDICT *dict= get_ndb()->getDictionary();
+  Ndb * ndb= get_ndb();
+  NDBDICT *dict= ndb->getDictionary();
   DBUG_ENTER("invalidate_dictionary_cache");
   DBUG_PRINT("info", ("invalidating %s", m_tabname));
 
@@ -459,6 +460,7 @@ void ha_ndbcluster::invalidate_dictionary_cache(bool global)
   }
   else
     dict->removeCachedTable(m_tabname);
+  build_index_list(ndb, table, ILBP_OPEN);
   table->s->version=0L;			/* Free when thread is ready */
   /* Invalidate indexes */
   for (uint i= 0; i < table->s->keys; i++)
@@ -470,17 +472,23 @@ void ha_ndbcluster::invalidate_dictionary_cache(bool global)
     switch (idx_type) {
     case PRIMARY_KEY_ORDERED_INDEX:
     case ORDERED_INDEX:
+      if (!index)
+        break;
       if (global)
         dict->invalidateIndex(index->getName(), m_tabname);
       else
         dict->removeCachedIndex(index->getName(), m_tabname);
       break;
     case UNIQUE_ORDERED_INDEX:
+      if (!index)
+        break;
       if (global)
         dict->invalidateIndex(index->getName(), m_tabname);
       else
         dict->removeCachedIndex(index->getName(), m_tabname);
     case UNIQUE_INDEX:
+      if (!unique_index)
+        break;
       if (global)
         dict->invalidateIndex(unique_index->getName(), m_tabname);
       else