Merge xiphis.org:/anubis/antony/work/mysql-5.0-engines

into  xiphis.org:/anubis/antony/work/merge.20080220/mysql-5.0

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/r/ndb_auto_increment.result

@@ -421,10 +421,10 @@ select * from t1 order by a;
 a
 1
 20
-21
 33
 34
 35
+65
 insert into t1 values (100);
 insert into t1 values (NULL);
 insert into t1 values (NULL);
@@ -432,11 +432,11 @@ select * from t1 order by a;
 a
 1
 20
-21
-22
 33
 34
 35
+65
+66
 100
 101
 set auto_increment_offset = @old_auto_increment_offset;

mysql-test/r/ndb_bug31477.result

+drop table if exists t1;
+create table t1(a int primary key, b int, c int, unique(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+begin;
+insert into t1 values (1,1,1);
+begin;
+update t1 set c = 2 where b = 1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+rollback;
+drop table t1;
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+begin;
+insert into t1 values (1,1,1);
+begin;
+update t1 set c = 2 where b = 1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+rollback;
+drop table t1;
+--con1
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (1,1,1);
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+begin;
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+--con1 c=30
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	30
+2	2	2
+3	3	3
+4	4	4
+--con2 c=1
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	1
+2	2	2
+3	3	3
+4	4	4
+--con1
+delete from t1 where a = 1;
+--con1 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+--con2 c=1
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	1
+2	2	2
+3	3	3
+4	4	4
+--con1
+commit;
+--con1 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+--con2 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+--con1
+begin;
+insert into t1 values (1,1,1);
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+--con1 c=30
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	30
+2	2	2
+3	3	3
+4	4	4
+--con2 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+drop table t1;

mysql-test/r/ndb_condition_pushdown.result

@@ -1904,6 +1904,12 @@ a	b	d
 10	1	4369
 20	2	8738
 50	5	21845
+-- big filter just below limit
+a	b	d
+10	1	4369
+20	2	8738
+50	5	21845
+-- big filter just above limit
 a	b	d
 10	1	4369
 20	2	8738

mysql-test/r/ndb_restore.result

@@ -266,21 +266,41 @@ a
 2000
 3000
 10000
-show table status like 't1_c';
-Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-X	X	X	X	X	X	X	X	X	X	3001	X	X	X	X	X	X	X
-show table status like 't2_c';
-Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-X	X	X	X	X	X	X	X	X	X	501	X	X	X	X	X	X	X
-show table status like 't4_c';
-Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-X	X	X	X	X	X	X	X	X	X	290000001	X	X	X	X	X	X	X
-show table status like 't7_c';
-Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-X	X	X	X	X	X	X	X	X	X	29	X	X	X	X	X	X	X
-show table status like 't10_c';
-Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-X	X	X	X	X	X	X	X	X	X	10001	X	X	X	X	X	X	X
+select max(capgoaledatta) from t1_c;
+max(capgoaledatta)
+3000
+select auto_increment from information_schema.tables
+where table_name = 't1_c';
+auto_increment
+3001
+select max(capgotod) from t2_c;
+max(capgotod)
+500
+select auto_increment from information_schema.tables
+where table_name = 't2_c';
+auto_increment
+501
+select max(capfa) from t4_c;
+max(capfa)
+290000000
+select auto_increment from information_schema.tables
+where table_name = 't4_c';
+auto_increment
+290000001
+select max(dardtestard) from t7_c;
+max(dardtestard)
+28
+select auto_increment from information_schema.tables
+where table_name = 't7_c';
+auto_increment
+29
+select max(a) from t10_c;
+max(a)
+10000
+select auto_increment from information_schema.tables
+where table_name = 't10_c';
+auto_increment
+10001
 drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9, t10;
 drop table if exists t1_c,t2_c,t3_c,t4_c,t5_c,t6_c,t7_c,t8_c,t9_c, t10_c;
 520093696,<the_backup_id>

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;

mysql-test/t/ndb_bug31477.test

+--source include/have_ndb.inc
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
+# setup
+
+connect (con1,localhost,root,,test);
+connect (con2,localhost,root,,test);
+
+# unique index
+connection con1;
+create table t1(a int primary key, b int, c int, unique(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+
+begin;
+insert into t1 values (1,1,1);
+
+connection con2;
+begin;
+--error 1205
+update t1 set c = 2 where b = 1;
+rollback;
+
+connection con1;
+rollback;
+drop table t1;
+# ordered index
+
+connection con1;
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+
+begin;
+insert into t1 values (1,1,1);
+
+connection con2;
+begin;
+--error 1205
+update t1 set c = 2 where b = 1;
+rollback;
+
+connection con1;
+rollback;
+drop table t1;
+
+# multiple versions
+
+--echo --con1
+connection con1;
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (1,1,1);
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+
+begin;
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+
+--echo --con1 c=30
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=1
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+--echo --con1
+connection con1;
+delete from t1 where a = 1;
+
+--echo --con1 c=none
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=1
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+--echo --con1
+connection con1;
+commit;
+
+--echo --con1 c=none
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=none
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+--echo --con1
+connection con1;
+begin;
+insert into t1 values (1,1,1);
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+
+--echo --con1 c=30
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=none
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+# this fails with "no such table" via con2 ???
+connection con1;
+drop table t1;

mysql-test/t/ndb_restore.test

@@ -231,16 +231,21 @@ select count(*)
 select * from t10_c order by a;
 # Bug #27775 cont'd
 # - auto inc info should be correct
---replace_column 1 X 2 X 3 X 4 X 5 X 6 X 7 X 8 X 9 X 10 X 12 X 13 X 14 X 15 X 16 X 17 X 18 X
-show table status like 't1_c';
---replace_column 1 X 2 X 3 X 4 X 5 X 6 X 7 X 8 X 9 X 10 X 12 X 13 X 14 X 15 X 16 X 17 X 18 X
-show table status like 't2_c';
---replace_column 1 X 2 X 3 X 4 X 5 X 6 X 7 X 8 X 9 X 10 X 12 X 13 X 14 X 15 X 16 X 17 X 18 X
-show table status like 't4_c';
---replace_column 1 X 2 X 3 X 4 X 5 X 6 X 7 X 8 X 9 X 10 X 12 X 13 X 14 X 15 X 16 X 17 X 18 X
-show table status like 't7_c';
---replace_column 1 X 2 X 3 X 4 X 5 X 6 X 7 X 8 X 9 X 10 X 12 X 13 X 14 X 15 X 16 X 17 X 18 X
-show table status like 't10_c';
+select max(capgoaledatta) from t1_c;
+select auto_increment from information_schema.tables
+where table_name = 't1_c';
+select max(capgotod) from t2_c;
+select auto_increment from information_schema.tables
+where table_name = 't2_c';
+select max(capfa) from t4_c;
+select auto_increment from information_schema.tables
+where table_name = 't4_c';
+select max(dardtestard) from t7_c;
+select auto_increment from information_schema.tables
+where table_name = 't7_c';
+select max(a) from t10_c;
+select auto_increment from information_schema.tables
+where table_name = 't10_c';
 
 --disable_warnings
 drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9, t10;

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/ndbapi/Ndb.hpp

@@ -1386,22 +1386,22 @@ public:
    *
    * @param cacheSize number of values to cache in this Ndb object
    *
-   * @return 0 or -1 on error, and tupleId in out parameter
+   * @return 0 or -1 on error, and autoValue in out parameter
    */
   int getAutoIncrementValue(const char* aTableName, 
-                            Uint64 & tupleId, Uint32 cacheSize, 
+                            Uint64 & autoValue, Uint32 cacheSize, 
 			    Uint64 step = 1, Uint64 start = 1);
   int getAutoIncrementValue(const NdbDictionary::Table * aTable, 
-                            Uint64 & tupleId, Uint32 cacheSize, 
+                            Uint64 & autoValue, Uint32 cacheSize, 
 			    Uint64 step = 1, Uint64 start = 1);
   int readAutoIncrementValue(const char* aTableName,
-                             Uint64 & tupleId);
+                             Uint64 & autoValue);
   int readAutoIncrementValue(const NdbDictionary::Table * aTable,
-                             Uint64 & tupleId);
+                             Uint64 & autoValue);
   int setAutoIncrementValue(const char* aTableName,
-                            Uint64 tupleId, bool increase);
+                            Uint64 autoValue, bool modify);
   int setAutoIncrementValue(const NdbDictionary::Table * aTable,
-                            Uint64 tupleId, bool increase);
+                            Uint64 autoValue, bool modify);
 private:
   int getTupleIdFromNdb(Ndb_local_table_info* info,
                         Uint64 & tupleId, Uint32 cacheSize,
@@ -1409,7 +1409,9 @@ private:
   int readTupleIdFromNdb(Ndb_local_table_info* info,
                          Uint64 & tupleId);
   int setTupleIdInNdb(Ndb_local_table_info* info,
-                      Uint64 tupleId, bool increase);
+                      Uint64 tupleId, bool modify);
+  int checkTupleIdInNdb(Ndb_local_table_info* info,
+                         Uint64 tupleId);
   int opTupleIdOnNdb(Ndb_local_table_info* info, Uint64 & opValue, Uint32 op);
 public:
 

ndb/include/ndbapi/ndbapi_limits.h

@@ -26,6 +26,7 @@
 #define NDB_MAX_TUPLE_SIZE (NDB_MAX_TUPLE_SIZE_IN_WORDS*4)
 #define NDB_MAX_ACTIVE_EVENTS 100
 
-#define NDB_MAX_SCANFILTER_SIZE_IN_WORDS 50000
+/* TUP ZATTR_BUFFER_SIZE 16384 (minus 1) minus place for getValue()s */
+#define NDB_MAX_SCANFILTER_SIZE_IN_WORDS (16384 - 1 - 1024)
 
 #endif

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/debugger/SignalLoggerManager.cpp

@@ -129,7 +129,7 @@ SignalLoggerManager::log(LogMode logMode, const char * params)
   const int count = getParameter(blocks, "BLOCK=", params);
   
   int cnt = 0;
-  if((count == 1 && blocks[0] == "ALL") ||
+  if((count == 1 && !strcmp(blocks[0], "ALL")) ||
      count == 0){
     
     for (int number = 0; number < NO_OF_BLOCKS; ++number){

ndb/src/common/debugger/signaldata/ScanTab.cpp

@@ -69,7 +69,7 @@ printSCANTABCONF(FILE * output, const Uint32 * theData, Uint32 len, Uint16 recei
 	  sig->transId1, sig->transId2);
 
   fprintf(output, " requestInfo: Eod: %d OpCount: %d\n", 
-	  (requestInfo & ScanTabConf::EndOfData == ScanTabConf::EndOfData),
+	  (requestInfo & ScanTabConf::EndOfData) == ScanTabConf::EndOfData,
 	  (requestInfo & (~ScanTabConf::EndOfData)));
   size_t op_count= requestInfo & (~ScanTabConf::EndOfData);
   if(op_count){

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/common/util/NdbOut.cpp

@@ -29,7 +29,7 @@ static const char * fms[] = {
   "%d", "0x%08x",      // Int32
   "%u", "0x%08x",      // Uint32
   "%lld", "0x%016llx", // Int64
-  "%llu", "0x%016llx"  // Uint64
+  "%llu", "0x%016llx", // Uint64
   "%llu", "0x%016llx"  // UintPtr
 };
 

ndb/src/kernel/blocks/backup/Backup.cpp

@@ -374,15 +374,17 @@ Backup::execCONTINUEB(Signal* signal)
     ptr.p->files.getPtr(filePtr, ptr.p->ctlFilePtr);
     FsBuffer & buf = filePtr.p->operation.dataBuffer;
     
-    if(buf.getFreeSize() + buf.getMinRead() < buf.getUsableSize()) {
+    if(buf.getFreeSize() < buf.getMaxWrite()) {
       jam();
       TablePtr tabPtr LINT_SET_PTR;
       c_tablePool.getPtr(tabPtr, Tdata2);
       
-      DEBUG_OUT("Backup - Buffer full - " << buf.getFreeSize()
-		<< " + " << buf.getMinRead()
-		<< " < " << buf.getUsableSize()
-		<< " - tableId = " << tabPtr.p->tableId);
+      DEBUG_OUT("Backup - Buffer full - " 
+                << buf.getFreeSize()
+		<< " < " << buf.getMaxWrite()
+                << " (sz: " << buf.getUsableSize()
+                << " getMinRead: " << buf.getMinRead()
+		<< ") - tableId = " << tabPtr.p->tableId);
       
       signal->theData[0] = BackupContinueB::BUFFER_FULL_META;
       signal->theData[1] = Tdata1;

ndb/src/kernel/blocks/backup/Backup.hpp

@@ -518,7 +518,7 @@ public:
   Uint32 m_diskless;
 
   STATIC_CONST(NO_OF_PAGES_META_FILE = 
-	       (MAX_WORDS_META_FILE + BACKUP_WORDS_PER_PAGE - 1) / 
+	       (2*MAX_WORDS_META_FILE + BACKUP_WORDS_PER_PAGE - 1) / 
 	       BACKUP_WORDS_PER_PAGE);
   
   /**

ndb/src/kernel/blocks/dbtup/Dbtup.hpp

@@ -198,6 +198,7 @@
 #define ZUNSUPPORTED_BRANCH 892
 
 #define ZSTORED_SEIZE_ATTRINBUFREC_ERROR 873 // Part of Scan
+#define ZSTORED_TOO_MUCH_ATTRINFO_ERROR 874
 
 #define ZREAD_ONLY_CONSTRAINT_VIOLATION 893
 #define ZVAR_SIZED_NOT_SUPPORTED 894
@@ -1085,7 +1086,7 @@ public:
   /*
    * TUX checks if tuple is visible to scan.
    */
-  bool tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, Uint32 savePointId);
+  bool tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, bool dirty, Uint32 savePointId);
 
 private:
   BLOCK_DEFINES(Dbtup);
@@ -1942,6 +1943,8 @@ private:
   bool getPageThroughSavePoint(Operationrec* const regOperPtr,
                                Operationrec* const leaderOpPtr);
 
+  bool find_savepoint(OperationrecPtr& loopOpPtr, Uint32 savepointId);
+
   Uint32 calculateChecksum(Page* const pagePtr, Uint32 tupHeadOffset, Uint32 tupHeadSize);
   void setChecksum(Page* const pagePtr, Uint32 tupHeadOffset, Uint32 tupHeadSize);
 
@@ -2171,7 +2174,8 @@ private:
                      Operationrec* regOperPtr,
                      Uint32 lenAttrInfo);
   void storedSeizeAttrinbufrecErrorLab(Signal* signal,
-                                       Operationrec* regOperPtr);
+                                       Operationrec* regOperPtr,
+                                       Uint32 errorCode);
   bool storedProcedureAttrInfo(Signal* signal,
                                Operationrec* regOperPtr,
                                Uint32 length,
@@ -2467,4 +2471,22 @@ bool Dbtup::isPageUndoLogged(Fragrecord* const regFragPtr,
   return false;
 }//Dbtup::isUndoLoggingNeeded()
 
+inline
+bool Dbtup::find_savepoint(OperationrecPtr& loopOpPtr, Uint32 savepointId)
+{
+  while (true) {
+    if (savepointId > loopOpPtr.p->savePointId) {
+      jam();
+      return true;
+    }
+    // note 5.0 has reversed next/prev pointers
+    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
+    if (loopOpPtr.i == RNIL) {
+      break;
+    }
+    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
+  }
+  return false;
+}
+
 #endif

ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp

@@ -77,6 +77,14 @@ void Dbtup::copyAttrinfo(Signal* signal,
     RbufLen = copyAttrBufPtr.p->attrbuf[ZBUF_DATA_LEN];
     Rnext = copyAttrBufPtr.p->attrbuf[ZBUF_NEXT];
     Rfirst = cfirstfreeAttrbufrec;
+    /*
+     * ATTRINFO comes from 2 mutually exclusive places:
+     * 1) TUPKEYREQ (also interpreted part)
+     * 2) STORED_PROCREQ before scan start
+     * Assert here that both have a check for overflow.
+     * The "<" instead of "<=" is intentional.
+     */
+    ndbrequire(RinBufIndex + RbufLen < ZATTR_BUFFER_SIZE);
     MEMCOPY_NO_WORDS(&inBuffer[RinBufIndex],
                      &copyAttrBufPtr.p->attrbuf[0],
                      RbufLen);

ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp

@@ -246,8 +246,18 @@ Dbtup::accReadPk(Uint32 tableId, Uint32 fragId, Uint32 fragPageId, Uint32 pageIn
   return ret;
 }
 
+/*
+ * TUX index contains all tuple versions.  A scan in TUX has scanned
+ * one of them and asks if it can be returned as scan result.  This
+ * depends on trans id, dirty read flag, and savepoint within trans.
+ *
+ * Previously this faked a ZREAD operation and used getPage().
+ * In TUP getPage() is run after ACC locking, but TUX comes here
+ * before ACC access.  Instead of modifying getPage() it is more
+ * clear to do the full check here.
+ */
 bool
-Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, Uint32 savePointId)
+Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, bool dirty, Uint32 savePointId)
 {
   ljamEntry();
   FragrecordPtr fragPtr;
@@ -256,33 +266,73 @@ Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 tra
   TablerecPtr tablePtr;
   tablePtr.i = fragPtr.p->fragTableId;
   ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
-  // get page
   PagePtr pagePtr;
-  Uint32 fragPageId = tupAddr >> MAX_TUPLES_BITS;
-  Uint32 pageIndex = tupAddr & ((1 << MAX_TUPLES_BITS ) - 1);
-  // use temp op rec
-  Operationrec tempOp;
-  tempOp.fragPageId = fragPageId;
-  tempOp.pageIndex = pageIndex;
-  tempOp.transid1 = transId1;
-  tempOp.transid2 = transId2;
-  tempOp.savePointId = savePointId;
-  tempOp.optype = ZREAD;
-  tempOp.dirtyOp = 1;
-  if (getPage(pagePtr, &tempOp, fragPtr.p, tablePtr.p)) {
-    /*
-    * We use the normal getPage which will return the tuple to be used
-    * for this transaction and savepoint id.  If its tuple version
-    * equals the requested then we have a visible tuple otherwise not.
-    */
-    ljam();
-    Uint32 read_tupVersion = pagePtr.p->pageWord[tempOp.pageOffset + 1];
-    if (read_tupVersion == tupVersion) {
+  pagePtr.i = pageId;
+  ptrCheckGuard(pagePtr, cnoOfPage, page);
+
+  OperationrecPtr currOpPtr;
+  currOpPtr.i = pagePtr.p->pageWord[pageOffset];
+  if (currOpPtr.i == RNIL) {
     ljam();
+    // tuple has no operation, any scan can see it
     return true;
   }
+  ptrCheckGuard(currOpPtr, cnoOfOprec, operationrec);
+
+  const bool sameTrans =
+    transId1 == currOpPtr.p->transid1 &&
+    transId2 == currOpPtr.p->transid2;
+
+  bool res = false;
+  OperationrecPtr loopOpPtr = currOpPtr;
+
+  if (!sameTrans) {
+    ljam();
+    if (!dirty) {
+      ljam();
+      if (currOpPtr.p->prevActiveOp == RNIL) {
+        ljam();
+        // last op - TUX makes ACC lock request in same timeslice
+        res = true;
+      }
     }
-  return false;
+    else {
+      // loop to first op (returns false)
+      find_savepoint(loopOpPtr, 0);
+      const Uint32 op_type = loopOpPtr.p->optype;
+
+      if (op_type != ZINSERT) {
+        ljam();
+        // read committed version from the page
+        const Uint32 origVersion = pagePtr.p->pageWord[pageOffset + 1];
+        if (origVersion == tupVersion) {
+          ljam();
+          res = true;
+        }
+      }
+    }
+  }
+  else {
+    ljam();
+    // for own trans, ignore dirty flag
+
+    if (find_savepoint(loopOpPtr, savePointId)) {
+      ljam();
+      const Uint32 op_type = loopOpPtr.p->optype;
+
+      if (op_type != ZDELETE) {
+        ljam();
+        // check if this op has produced the scanned version
+        Uint32 loopVersion = loopOpPtr.p->tupVersion;
+        if (loopVersion == tupVersion) {
+          ljam();
+          res = true;
+        }
+      }
+    }
+  }
+
+  return res;
 }
 
 // ordered index build

ndb/src/kernel/blocks/dbtup/DbtupStoredProcDef.cpp

@@ -108,6 +108,11 @@ void Dbtup::scanProcedure(Signal* signal,
   regOperPtr->attrinbufLen = lenAttrInfo;
   regOperPtr->currentAttrinbufLen = 0;
   regOperPtr->pageOffset = storedPtr.i;
+  if (lenAttrInfo >= ZATTR_BUFFER_SIZE) { // yes ">="
+    jam();
+    // send REF and change state to ignore the ATTRINFO to come
+    storedSeizeAttrinbufrecErrorLab(signal, regOperPtr, ZSTORED_TOO_MUCH_ATTRINFO_ERROR);
+  }
 }//Dbtup::scanProcedure()
 
 void Dbtup::copyProcedure(Signal* signal,
@@ -146,7 +151,7 @@ bool Dbtup::storedProcedureAttrInfo(Signal* signal,
   Uint32 RnoFree = cnoFreeAttrbufrec;
   if (ERROR_INSERTED(4004) && !copyProcedure) {
     CLEAR_ERROR_INSERT_VALUE;
-    storedSeizeAttrinbufrecErrorLab(signal, regOperPtr);
+    storedSeizeAttrinbufrecErrorLab(signal, regOperPtr, ZSTORED_SEIZE_ATTRINBUFREC_ERROR);
     return false;
   }//if
   regOperPtr->currentAttrinbufLen += length;
@@ -162,7 +167,7 @@ bool Dbtup::storedProcedureAttrInfo(Signal* signal,
     regAttrPtr.p->attrbuf[ZBUF_NEXT] = RNIL;
   } else {
     ljam();
-    storedSeizeAttrinbufrecErrorLab(signal, regOperPtr);
+    storedSeizeAttrinbufrecErrorLab(signal, regOperPtr, ZSTORED_SEIZE_ATTRINBUFREC_ERROR);
     return false;
   }//if
   if (regOperPtr->firstAttrinbufrec == RNIL) {
@@ -190,7 +195,7 @@ bool Dbtup::storedProcedureAttrInfo(Signal* signal,
   }//if
   if (ERROR_INSERTED(4005) && !copyProcedure) {
     CLEAR_ERROR_INSERT_VALUE;
-    storedSeizeAttrinbufrecErrorLab(signal, regOperPtr);
+    storedSeizeAttrinbufrecErrorLab(signal, regOperPtr, ZSTORED_SEIZE_ATTRINBUFREC_ERROR);
     return false;
   }//if
 
@@ -211,7 +216,8 @@ bool Dbtup::storedProcedureAttrInfo(Signal* signal,
 }//Dbtup::storedProcedureAttrInfo()
 
 void Dbtup::storedSeizeAttrinbufrecErrorLab(Signal* signal,
-                                            Operationrec* regOperPtr)
+                                            Operationrec* regOperPtr,
+                                            Uint32 errorCode)
 {
   StoredProcPtr storedPtr;
   c_storedProcPool.getPtr(storedPtr, (Uint32)regOperPtr->pageOffset);
@@ -222,7 +228,7 @@ void Dbtup::storedSeizeAttrinbufrecErrorLab(Signal* signal,
   regOperPtr->lastAttrinbufrec = RNIL;
   regOperPtr->transstate = ERROR_WAIT_STORED_PROCREQ;
   signal->theData[0] = regOperPtr->userpointer;
-  signal->theData[1] = ZSTORED_SEIZE_ATTRINBUFREC_ERROR;
+  signal->theData[1] = errorCode;
   signal->theData[2] = regOperPtr->pageOffset;
   sendSignal(regOperPtr->userblockref, GSN_STORED_PROCREF, signal, 3, JBB);
 }//Dbtup::storedSeizeAttrinbufrecErrorLab()

ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp

@@ -984,7 +984,8 @@ Dbtux::scanVisible(ScanOpPtr scanPtr, TreeEnt ent)
   const Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
   Uint32 fragBit = ent.m_fragBit;
   Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[fragBit];
-  Uint32 tupAddr = getTupAddr(frag, ent);
+  Uint32 pageId = ent.m_tupLoc.getPageId();
+  Uint32 pageOffset = ent.m_tupLoc.getPageOffset();
   Uint32 tupVersion = ent.m_tupVersion;
   // check for same tuple twice in row
   if (scan.m_scanEnt.m_tupLoc == ent.m_tupLoc &&
@@ -994,8 +995,9 @@ Dbtux::scanVisible(ScanOpPtr scanPtr, TreeEnt ent)
   }
   Uint32 transId1 = scan.m_transId1;
   Uint32 transId2 = scan.m_transId2;
+  bool dirty = scan.m_readCommitted;
   Uint32 savePointId = scan.m_savePointId;
-  bool ret = c_tup->tuxQueryTh(tableFragPtrI, tupAddr, tupVersion, transId1, transId2, savePointId);
+  bool ret = c_tup->tuxQueryTh(tableFragPtrI, pageId, pageOffset, tupVersion, transId1, transId2, dirty, savePointId);
   jamEntry();
   return ret;
 }

ndb/src/kernel/vm/DLHashTable.hpp

@@ -45,8 +45,8 @@ public:
   /**
    * Seize element from pool - return i
    *
-   * Note must be either added using <b>add</b> or released 
-   * using <b>release</b>
+   * Note *must* be added using <b>add</b> (even before hash.release)
+   *             or be released using pool
    */
   bool seize(Ptr<T> &);
 
@@ -374,8 +374,15 @@ DLHashTable<T>::remove(Ptr<T> & ptr){
     prevP->nextHash = next;
   } else {
     const Uint32 hv = ptr.p->hashValue() & mask;  
+    if (hashValues[hv] == ptr.i)
+    {
       hashValues[hv] = next;
     }
+    else
+    {
+      // Will add assert in 5.1
+    }
+  }
   
   if(next != RNIL){
     T * nextP = thePool.getPtr(next);
@@ -395,8 +402,15 @@ DLHashTable<T>::release(Ptr<T> & ptr){
     prevP->nextHash = next;
   } else {
     const Uint32 hv = ptr.p->hashValue() & mask;  
+    if (hashValues[hv] == ptr.i)
+    {
       hashValues[hv] = next;
     }
+    else
+    {
+      // Will add assert in 5.1
+    }
+  }
   
   if(next != RNIL){
     T * nextP = thePool.getPtr(next);

ndb/src/kernel/vm/DLHashTable2.hpp

@@ -43,8 +43,8 @@ public:
   /**
    * Seize element from pool - return i
    *
-   * Note must be either added using <b>add</b> or released 
-   * using <b>release</b>
+   * Note *must* be added using <b>add</b> (even before hash.release)
+   *             or be released using pool
    */
   bool seize(Ptr<T> &);
 
@@ -375,8 +375,15 @@ DLHashTable2<T, U>::remove(Ptr<T> & ptr){
     prevP->nextHash = next;
   } else {
     const Uint32 hv = ptr.p->hashValue() & mask;  
+    if (hashValues[hv] == ptr.i)
+    {
       hashValues[hv] = next;
     }
+    else
+    {
+      // Will add assert in 5.1
+    }
+  }
   
   if(next != RNIL){
     T * nextP = (T*)thePool.getPtr(next);       // cast
@@ -396,8 +403,15 @@ DLHashTable2<T, U>::release(Ptr<T> & ptr){
     prevP->nextHash = next;
   } else {
     const Uint32 hv = ptr.p->hashValue() & mask;  
+    if (hashValues[hv] == ptr.i)
+    {
       hashValues[hv] = next;
     }
+    else
+    {
+      // Will add assert in 5.1
+    }
+  }
   
   if(next != RNIL){
     T * nextP = (T*)thePool.getPtr(next);       // cast

ndb/src/kernel/vm/pc.hpp

@@ -49,7 +49,7 @@
   theEmulatedJamBlockNumber = number(); \
   Uint32 tEmulatedJamIndex = theEmulatedJamIndex; \
   *(Uint32*)(theEmulatedJam + tEmulatedJamIndex) = \
-    ((theEmulatedJamBlockNumber << 20) | line); \
+    ((theEmulatedJamBlockNumber << 20) | (line)); \
   theEmulatedJamIndex = (tEmulatedJamIndex + 4) & JAM_MASK; }
 
 #else
@@ -72,7 +72,7 @@
   theEmulatedJamBlockNumber = number(); \
   Uint32 tEmulatedJamIndex = theEmulatedJamIndex; \
   *(Uint32*)((UintPtr)theEmulatedJam + (Uint32)tEmulatedJamIndex) = \
-    ((theEmulatedJamBlockNumber << 20) | line); \
+    ((theEmulatedJamBlockNumber << 20) | (line)); \
   theEmulatedJamIndex = (tEmulatedJamIndex + 4) & JAM_MASK; }
 
 #endif
@@ -232,7 +232,7 @@
 #define MEMCOPY_PAGE(to, from, page_size_in_bytes) \
   memcpy((void*)(to), (void*)(from), (size_t)(page_size_in_bytes));
 #define MEMCOPY_NO_WORDS(to, from, no_of_words) \
-  memcpy((to), (void*)(from), (size_t)(no_of_words << 2));
+  memcpy((to), (void*)(from), (size_t)((no_of_words) << 2));
 
 template <class T>
 struct Ptr {

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/src/ndbapi/DictCache.cpp

@@ -46,6 +46,7 @@ Ndb_local_table_info::Ndb_local_table_info(NdbTableImpl *table_impl)
   m_table_impl= table_impl;
   m_first_tuple_id = ~(Uint64)0;
   m_last_tuple_id = ~(Uint64)0;
+  m_highest_seen = 0;
 }
 
 Ndb_local_table_info::~Ndb_local_table_info()

ndb/src/ndbapi/DictCache.hpp

@@ -36,6 +36,7 @@ public:
   // range of cached tuple ids per thread
   Uint64 m_first_tuple_id;
   Uint64 m_last_tuple_id;
+  Uint64 m_highest_seen;
 
   Uint64 m_local_data[1]; // Must be last member. Used to access extra space.
 private:

ndb/src/ndbapi/Ndb.cpp

@@ -767,8 +767,8 @@ Ndb::getNodeId()
 }
 
 /****************************************************************************
-Uint64 getAutoIncrementValue( const char* aTableName,
-                              Uint64 & tupleId, 
+int    getAutoIncrementValue( const char* aTableName,
+                              Uint64 & autoValue, 
                               Uint32 cacheSize, 
                               Uint64 step,
                               Uint64 start);
@@ -779,6 +779,7 @@ Parameters:     aTableName (IN) : The table name.
                 step       (IN) : Specifies the step between the 
                                   autoincrement values.
                 start      (IN) : Start value for first value
+Returns:        0 if succesful, -1 if error encountered
 Remark:		Returns a new autoincrement value to the application.
                 The autoincrement values can be increased by steps
                 (default 1) and a number of values can be prefetched
@@ -892,9 +893,18 @@ Ndb::getTupleIdFromNdb(Ndb_local_table_info* info,
   DBUG_RETURN(0);
 }
 
+/****************************************************************************
+int readAutoIncrementValue( const char* aTableName,
+                            Uint64 & autoValue);
+
+Parameters:     aTableName (IN) : The table name.
+                autoValue  (OUT) : The current autoincrement value
+Returns:        0 if succesful, -1 if error encountered
+Remark:		Returns the current autoincrement value to the application.
+****************************************************************************/
 int
 Ndb::readAutoIncrementValue(const char* aTableName,
-                            Uint64 & tupleId)
+                            Uint64 & autoValue)
 {
   DBUG_ENTER("Ndb::readAutoIncrementValue");
   BaseString internal_tabname(internalize_table_name(aTableName));
@@ -905,15 +915,15 @@ Ndb::readAutoIncrementValue(const char* aTableName,
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(-1);
   }
-  if (readTupleIdFromNdb(info, tupleId) == -1)
+  if (readTupleIdFromNdb(info, autoValue) == -1)
     DBUG_RETURN(-1);
-  DBUG_PRINT("info", ("value %lu", (ulong)tupleId));
+  DBUG_PRINT("info", ("value %lu", (ulong)autoValue));
   DBUG_RETURN(0);
 }
 
 int
 Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable,
-                            Uint64 & tupleId)
+                            Uint64 & autoValue)
 {
   DBUG_ENTER("Ndb::readAutoIncrementValue");
   assert(aTable != 0);
@@ -926,9 +936,9 @@ Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable,
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(-1);
   }
-  if (readTupleIdFromNdb(info, tupleId) == -1)
+  if (readTupleIdFromNdb(info, autoValue) == -1)
     DBUG_RETURN(-1);
-  DBUG_PRINT("info", ("value %lu", (ulong)tupleId));
+  DBUG_PRINT("info", ("value %lu", (ulong)autoValue));
   DBUG_RETURN(0);
 }
 
@@ -956,9 +966,20 @@ Ndb::readTupleIdFromNdb(Ndb_local_table_info* info,
   DBUG_RETURN(0);
 }
 
+/****************************************************************************
+int setAutoIncrementValue( const char* aTableName,
+                           Uint64 autoValue,
+                           bool modify);
+
+Parameters:     aTableName (IN) : The table name.
+                autoValue  (IN) : The new autoincrement value
+                modify     (IN) : Modify existing value (not initialization)
+Returns:        0 if succesful, -1 if error encountered
+Remark:		Sets a new autoincrement value for the application.
+****************************************************************************/
 int
 Ndb::setAutoIncrementValue(const char* aTableName,
-                           Uint64 tupleId, bool increase)
+                           Uint64 autoValue, bool modify)
 {
   DBUG_ENTER("Ndb::setAutoIncrementValue");
   BaseString internal_tabname(internalize_table_name(aTableName));
@@ -969,14 +990,14 @@ Ndb::setAutoIncrementValue(const char* aTableName,
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(-1);
   }
-  if (setTupleIdInNdb(info, tupleId, increase) == -1)
+  if (setTupleIdInNdb(info, autoValue, modify) == -1)
     DBUG_RETURN(-1);
   DBUG_RETURN(0);
 }
 
 int
 Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable,
-                           Uint64 tupleId, bool increase)
+                           Uint64 autoValue, bool modify)
 {
   DBUG_ENTER("Ndb::setAutoIncrementValue");
   assert(aTable != 0);
@@ -989,17 +1010,19 @@ Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable,
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(-1);
   }
-  if (setTupleIdInNdb(info, tupleId, increase) == -1)
+  if (setTupleIdInNdb(info, autoValue, modify) == -1)
     DBUG_RETURN(-1);
   DBUG_RETURN(0);
 }
 
 int
 Ndb::setTupleIdInNdb(Ndb_local_table_info* info,
-                     Uint64 tupleId, bool increase)
+                     Uint64 tupleId, bool modify)
 {
   DBUG_ENTER("Ndb::setTupleIdInNdb");
-  if (increase)
+  if (modify)
+  {
+    if (checkTupleIdInNdb(info, tupleId))
     {
       if (info->m_first_tuple_id != info->m_last_tuple_id)
       {
@@ -1019,9 +1042,11 @@ Ndb::setTupleIdInNdb(Ndb_local_table_info* info,
        * if tupleId <= NEXTID, do nothing.  otherwise update NEXTID to
        * tupleId and set cached range to first = last = tupleId - 1.
        */
-    if (opTupleIdOnNdb(info, tupleId, 2) == -1)
+      Uint64 opValue = tupleId;
+      if (opTupleIdOnNdb(info, opValue, 2) == -1)
         DBUG_RETURN(-1);
     }
+  }
   else
   {
     /*
@@ -1033,6 +1058,32 @@ Ndb::setTupleIdInNdb(Ndb_local_table_info* info,
   DBUG_RETURN(0);
 }
 
+int
+Ndb::checkTupleIdInNdb(Ndb_local_table_info* info, Uint64 tupleId)
+{
+  DBUG_ENTER("Ndb::checkTupleIdIndNdb");
+  if ((info->m_first_tuple_id != ~(Uint64)0) &&
+      (info->m_first_tuple_id > tupleId))
+  {
+   /*
+    * If we have ever cached a value in this object and this cached
+    * value is larger than the value we're trying to set then we
+    * need not check with the real value in the SYSTAB_0 table.
+    */
+    DBUG_RETURN(0);
+  }
+  if (info->m_highest_seen > tupleId)
+  {
+    /*
+     * Although we've never cached any higher value we have read
+     * a higher value and again it isn't necessary to change the
+     * auto increment value.
+     */
+    DBUG_RETURN(0);
+  }
+  DBUG_RETURN(1);
+}
+
 int
 Ndb::opTupleIdOnNdb(Ndb_local_table_info* info, Uint64 & opValue, Uint32 op)
 {
@@ -1094,6 +1145,7 @@ Ndb::opTupleIdOnNdb(Ndb_local_table_info* info, Uint64 & opValue, Uint32 op)
 
       info->m_first_tuple_id = ~(Uint64)0;
       info->m_last_tuple_id  = ~(Uint64)0;
+      info->m_highest_seen = 0;
       break;
     case 2:
       tOperation->interpretedUpdateTuple();
@@ -1103,19 +1155,18 @@ Ndb::opTupleIdOnNdb(Ndb_local_table_info* info, Uint64 & opValue, Uint32 op)
       // compare NEXTID >= opValue
       tOperation->branch_le(2, 1, 0);
       tOperation->write_attr("NEXTID", 1);
-      tOperation->interpret_exit_ok();
       tOperation->def_label(0);
-      tOperation->interpret_exit_nok(9999);
-      
+      tOperation->interpret_exit_ok();
+      tRecAttrResult = tOperation->getValue("NEXTID");
       if (tConnection->execute( Commit ) == -1)
       {
-        if (tConnection->theError.code != 9999)
         goto error_handler;
       }
       else
       {
+	info->m_highest_seen = tRecAttrResult->u_64_value();
 	DBUG_PRINT("info", 
-		   ("Setting next auto increment value (db) to %lu",
+		   ("Setting auto increment value (db) to %lu",
 		    (ulong)opValue));  
         info->m_first_tuple_id = info->m_last_tuple_id = opValue - 1;
       }
@@ -1126,7 +1177,7 @@ Ndb::opTupleIdOnNdb(Ndb_local_table_info* info, Uint64 & opValue, Uint32 op)
       tRecAttrResult = tOperation->getValue("NEXTID");
       if (tConnection->execute( Commit ) == -1 )
         goto error_handler;
-      opValue = tRecAttrResult->u_64_value(); // out
+      info->m_highest_seen = opValue = tRecAttrResult->u_64_value(); // out
       break;
     default:
       goto error_handler;
@@ -1415,11 +1466,7 @@ Ndb::printState(const char* fmt, ...)
   NdbMutex_Lock(ndb_print_state_mutex);
   bool dups = false;
   unsigned i;
-  ndbout << buf << " ndb=" << hex << this << dec;
-#ifndef NDB_WIN32
-  ndbout << " thread=" << (int)pthread_self();
-#endif
-  ndbout << endl;
+  ndbout << buf << " ndb=" << hex << (void*)this << endl;
   for (unsigned n = 0; n < MAX_NDB_NODES; n++) {
     NdbTransaction* con = theConnectionArray[n];
     if (con != 0) {

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -35,6 +35,7 @@
 #include "NdbBlobImpl.hpp"
 #include <AttributeHeader.hpp>
 #include <my_sys.h>
+#include <NdbSleep.h>
 
 #define DEBUG_PRINT 0
 #define INCOMPATIBLE_VERSION -2
@@ -886,7 +887,23 @@ NdbDictInterface::dictSignal(NdbApiSignal* signal,
 {
   DBUG_ENTER("NdbDictInterface::dictSignal");
   DBUG_PRINT("enter", ("useMasterNodeId: %d", useMasterNodeId));
-  for(Uint32 i = 0; i<RETRIES; i++){
+
+  int sleep = 50;
+  int mod = 5;
+
+  for(Uint32 i = 0; i<RETRIES; i++)
+  {
+    if (i > 0)
+      NdbSleep_MilliSleep(sleep + 10 * (rand() % mod));
+    if (i == RETRIES / 2)
+    {
+      mod = 10;
+    }
+    if (i == 3*RETRIES/4)
+    {
+      sleep = 100;
+    }
+
     //if (useMasterNodeId == 0)
     m_buffer.clear();
 

ndb/src/ndbapi/ndberror.c

@@ -291,6 +291,7 @@ ErrorBundle ErrorCodes[] = {
   { 242,  AE, "Zero concurrency in scan"},
   { 244,  AE, "Too high concurrency in scan"},
   { 269,  AE, "No condition and attributes to read in scan"},
+  { 874,  AE, "Too much attrinfo (e.g. scan filter) for scan in tuple manager" },
   { 4600, AE, "Transaction is already started"},
   { 4601, AE, "Transaction is not started"},
   { 4602, AE, "You must call getNdbOperation before executeScan" },

ndb/test/ndbapi/testInterpreter.cpp

@@ -77,6 +77,11 @@ int runTestIncValue32(NDBT_Context* ctx, NDBT_Step* step){
   const NdbDictionary::Table * pTab = ctx->getTab();
   Ndb* pNdb = GETNDB(step);
 
+  if (strcmp(pTab->getName(), "T1") != 0) {
+    g_err << "runTestBug19537: skip, table != T1" << endl;
+    return NDBT_OK;
+  }
+
 
   NdbConnection* pTrans = pNdb->startTransaction();
   if (pTrans == NULL){
@@ -258,6 +263,84 @@ int runTestBug19537(NDBT_Context* ctx, NDBT_Step* step){
 }
 
 
+int runTestBug34107(NDBT_Context* ctx, NDBT_Step* step){
+  int result = NDBT_OK;
+  const NdbDictionary::Table * pTab = ctx->getTab();
+  Ndb* pNdb = GETNDB(step);
+
+  int i;
+  for (i = 0; i <= 1; i++) {
+    g_info << "bug34107:" << (i == 0 ? " small" : " too big") << endl;
+
+    NdbConnection* pTrans = pNdb->startTransaction();
+    if (pTrans == NULL){
+      ERR(pNdb->getNdbError());
+      return NDBT_FAILED;
+    }
+    
+    NdbScanOperation* pOp = pTrans->getNdbScanOperation(pTab->getName());
+    if (pOp == NULL) {
+      ERR(pTrans->getNdbError());
+      pNdb->closeTransaction(pTrans);
+      return NDBT_FAILED;
+    }
+    
+    if (pOp->readTuples() == -1) {
+      ERR(pOp->getNdbError());
+      pNdb->closeTransaction(pTrans);
+      return NDBT_FAILED;
+    }
+
+    int n = i == 0 ? 10000 : 30000;
+    int k;
+
+    for (k = 0; k < n; k++) {
+
+      // inserts 1 word ATTRINFO
+
+      if (pOp->interpret_exit_ok() == -1) {
+        ERR(pOp->getNdbError());
+        pNdb->closeTransaction(pTrans);
+        return NDBT_FAILED;
+      }
+    }
+      
+    if (pTrans->execute(NoCommit) == -1) {
+      ERR(pTrans->getNdbError());
+      pNdb->closeTransaction(pTrans);
+      return NDBT_FAILED;
+    }
+
+    int ret;
+    while ((ret = pOp->nextResult()) == 0)
+      ;
+    g_info << "ret=" << ret << " err=" << pOp->getNdbError().code << endl;
+
+    if (i == 0 && ret != 1) {
+      ERR(pTrans->getNdbError());
+      pNdb->closeTransaction(pTrans);
+      return NDBT_FAILED;
+    }
+
+    if (i == 1 && ret != -1) {
+      g_err << "unexpected big filter success" << endl;
+      pNdb->closeTransaction(pTrans);
+      return NDBT_FAILED;
+    }
+    if (i == 1 && pOp->getNdbError().code != 874) {
+      g_err << "unexpected big filter error code, wanted 874" << endl;
+      ERR(pTrans->getNdbError());
+      pNdb->closeTransaction(pTrans);
+      return NDBT_FAILED;
+    }
+
+    pNdb->closeTransaction(pTrans);
+  }
+
+  return NDBT_OK;
+}
+
+
 NDBT_TESTSUITE(testInterpreter);
 TESTCASE("IncValue32", 
 	 "Test incValue for 32 bit integer\n"){ 
@@ -277,6 +360,12 @@ TESTCASE("Bug19537",
   INITIALIZER(runTestBug19537);
   FINALIZER(runClearTable);
 }
+TESTCASE("Bug34107",
+         "Test too big scan filter (error 874)\n"){
+  INITIALIZER(runLoadTable);
+  INITIALIZER(runTestBug34107);
+  FINALIZER(runClearTable);
+}
 #if 0
 TESTCASE("MaxTransactions", 
 	 "Start transactions until no more can be created\n"){ 

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

@@ -649,10 +649,10 @@ max-time: 1000
 cmd: testNdbApi
 args: -n Bug28443
 
-#max-time: 500
-#cmd: testInterpreter
-#args: T1 
-#
+max-time: 500
+cmd: testInterpreter
+args: T1 
+
 max-time: 150000
 cmd: testOperations
 args:

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