Merge mysql.com:/home/jonas/src/mysql-5.0

into mysql.com:/home/jonas/src/mysql-5.0-ndb

BitKeeper/etc/gone

@@ -457,6 +457,40 @@ ccarkner@nslinuxw10.bedford.progress.com|mysql-test/r/isolation.result|200103271
 ccarkner@nslinuxw10.bedford.progress.com|mysql-test/t/isolation.test|20010327145543|39049|6a39e4138dd4a456
 jani@hynda.mysql.fi|client/mysqlcheck|20010419221207|26716|363e3278166d84ec
 jcole@tetra.bedford.progress.com|BitKeeper/etc/logging_ok|20001004201211|30554
+magnus@neptunus.(none)|ndb/tools/ndbnet/Makefile.PL|20040414082442|56473|81be90388548652f
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net.pm|20040414082442|02828|425c84165071d5f6
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Base.pm|20040414082442|11671|50a6f0d38fa1a57c
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Client.pm|20040414082442|14568|45adb527c5cfdaf6
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Command.pm|20040414082442|17582|a20dff1bfd06ea7a
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Config.pm|20040414082442|20612|24456d4beba19604
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Database.pm|20040414082442|23641|cc02e455b02c557
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Env.pm|20040414082442|26624|ffeca4ab253818c3
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Node.pm|20040414082442|29671|276106bf4000fe2b
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/NodeApi.pm|20040414082442|32765|3abf1d7bc947d3d0
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/NodeDb.pm|20040414082443|01346|9e3055381da679ce
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/NodeMgmt.pm|20040414082443|04349|5123fbffcfce73cb
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/Server.pm|20040414082443|07453|70208c964371fd4d
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/ServerINET.pm|20040414082443|10474|fc72d28fedd17795
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Net/ServerUNIX.pm|20040414082443|13573|ad3e25435fceb8d0
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Run.pm|20040414082442|05790|9b414f2b4b76a4da
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Run/Base.pm|20040414082443|16656|dd382c9eec2c38b1
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Run/Database.pm|20040414082443|19701|ec533cd5dd6a54a2
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Run/Env.pm|20040414082443|22716|a343841fae6d0514
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Run/Node.pm|20040414082443|25798|339139ded5fcf795
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util.pm|20040414082442|08694|9be64f1195ec502d
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/Base.pm|20040414082443|28821|f39906f304e8a49
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/Dir.pm|20040414082443|31912|4cbe8dd8c57149a7
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/Event.pm|20040414082443|35023|f445c1c094f57cf
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/File.pm|20040414082443|38178|48da8e977f39bf7
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/IO.pm|20040414082443|41286|6c6a63ce63fe2ee5
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/Lock.pm|20040414082443|44445|366e56786358b3d8
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/Log.pm|20040414082443|47535|e0da9468a6f9213a
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/Socket.pm|20040414082443|50710|371ccd6c5a818fa0
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/SocketINET.pm|20040414082443|53816|db8a45b1360b69f2
+magnus@neptunus.(none)|ndb/tools/ndbnet/lib/NDB/Util/SocketUNIX.pm|20040414082443|56977|9ffab2ada7f61530
+magnus@neptunus.(none)|ndb/tools/ndbnet/ndbnet.pl|20040414082442|59371|8f8f1b18f0458269
+magnus@neptunus.(none)|ndb/tools/ndbnet/ndbnetd.pl|20040414082442|62363|e44f1cfe7e8bbe14
+magnus@neptunus.(none)|ndb/tools/ndbnet/ndbrun|20040414082442|65454|5db549d7436ac81b
 miguel@hegel.local|zlib/ChangeLog|20020319032513|28917|5d5425fc84737083
 miguel@hegel.local|zlib/Make_vms.com|20020319032513|57151|35050a50ec612bbf
 miguel@hegel.local|zlib/Makefile.riscos|20020319032513|63798|8ab53f195fe429af

ndb/include/kernel/signaldata/DictTabInfo.hpp

@@ -230,6 +230,7 @@ public:
   STATIC_CONST( StringType = 3 );
   
   // AttributeSize constants
+  STATIC_CONST( aBit = 0 );
   STATIC_CONST( an8Bit = 3 );
   STATIC_CONST( a16Bit = 4 );
   STATIC_CONST( a32Bit = 5 );
@@ -310,7 +311,8 @@ public:
     ExtDatetime = NdbSqlUtil::Type::Datetime,
     ExtTimespec = NdbSqlUtil::Type::Timespec,
     ExtBlob = NdbSqlUtil::Type::Blob,
-    ExtText = NdbSqlUtil::Type::Text
+    ExtText = NdbSqlUtil::Type::Text,
+    ExtBit = NdbSqlUtil::Type::Bit
   };
 
   // Attribute data interpretation
@@ -440,7 +442,13 @@ public:
         // head + inline part [ attr precision lower half ]
         AttributeArraySize = (NDB_BLOB_HEAD_SIZE << 2) + (AttributeExtPrecision & 0xFFFF);
         return true;
+      case DictTabInfo::ExtBit:
+	AttributeType = DictTabInfo::UnSignedType;
+	AttributeSize = DictTabInfo::aBit;
+	AttributeArraySize = AttributeExtLength;
+	return true;
       };
+      
       return false;
     }
     

ndb/include/ndbapi/NdbDictionary.hpp

@@ -187,7 +187,8 @@ public:
       Datetime,    ///< Precision down to 1 sec (sizeof(Datetime) == 8 bytes )
       Timespec,    ///< Precision down to 1 nsec(sizeof(Datetime) == 12 bytes )
       Blob,        ///< Binary large object (see NdbBlob)
-      Text         ///< Text blob
+      Text,         ///< Text blob,
+      Bit           ///< Bit, length specifies no of bits
     };
 
     /** 

ndb/include/util/Bitmask.hpp

@@ -131,10 +131,26 @@ public:
   static void setField(unsigned size, Uint32 data[],
       unsigned pos, unsigned len, Uint32 val);
 
+
+  /**
+   * getField - Get bitfield at given position and length
+   */
+  static void getField(unsigned size, const Uint32 data[],
+		       unsigned pos, unsigned len, Uint32 dst[]);
+  
+  /**
+   * setField - Set bitfield at given position and length
+   */
+  static void setField(unsigned size, Uint32 data[],
+		       unsigned pos, unsigned len, const Uint32 src[]);
+  
   /**
    * getText - Return as hex-digits (only for debug routines).
    */
   static char* getText(unsigned size, const Uint32 data[], char* buf);
+private:
+  static void getFieldImpl(const Uint32 data[], unsigned, unsigned, Uint32 []);
+  static void setFieldImpl(Uint32 data[], unsigned, unsigned, const Uint32 []);
 };
 
 inline bool
@@ -794,4 +810,38 @@ public:
   Bitmask() { this->clear();}
 };
 
+inline void
+BitmaskImpl::getField(unsigned size, const Uint32 data[],
+		      unsigned pos, unsigned len, Uint32 dst[])
+{
+  assert(len > 0);
+  assert(pos + len < (size << 5));
+  Uint32 word = pos >> 5;
+  Uint32 offset = pos & 31;
+  if(offset + len <= 32)
+  {
+    dst[0] = (data[word] >> offset) & ((1 << len) - 1);
+    return;
+  }
+  getFieldImpl(data, pos, len, dst);
+}
+
+inline void
+BitmaskImpl::setField(unsigned size, Uint32 data[],
+		      unsigned pos, unsigned len, const Uint32 src[])
+{
+  assert(len > 0);
+  assert(pos + len < (size << 5));
+  Uint32 word = pos >> 5;
+  Uint32 offset = pos & 31;
+  Uint32 mask = ((1 << len) - 1) << offset;
+  data[word] = (data[word] & ~mask) | ((src[0] << offset) & mask);
+  if(offset + len <= 32)
+  {
+    return;
+  }
+  setFieldImpl(data, pos, len, src);
+}
+
+
 #endif

ndb/include/util/NdbSqlUtil.hpp

@@ -83,7 +83,8 @@ public:
       Datetime,         // Precision down to 1 sec  (size 8 bytes)
       Timespec,         // Precision down to 1 nsec (size 12 bytes)
       Blob,             // Blob
-      Text              // Text blob
+      Text,             // Text blob,
+      Bit               // A bit
     };
     Enum m_typeId;
     Cmp* m_cmp;         // comparison method

ndb/src/common/util/Bitmask.cpp

+#include <Bitmask.hpp>
+#include <NdbOut.hpp>
+
+#ifndef __TEST_BITMASK__
+void
+BitmaskImpl::getFieldImpl(const Uint32 data[],
+			  unsigned pos, unsigned l, Uint32 dst[])
+{
+  Uint32 word;
+  Uint32 offset;
+  int next_offset,i;
+  int len= l;
+  for(i=0,next_offset=0;len >0;i++)
+  {
+    word = pos >> 5;
+    offset = pos & 31;
+
+    if(i%32==0)
+      dst[i/32]=0;
+
+    if(!next_offset && (offset+len) > 32)
+    {
+      dst[i/32] = (data[word] >> offset) & ((1 << (32-offset)) - 1);
+      next_offset = 32-offset;
+    }
+    else
+    {
+      dst[i/32]|= ((data[word] >> offset) & ((1 << len) - 1)) << next_offset;
+      next_offset = 0;
+    }
+    
+    if(len < 32-offset)
+      break;
+    
+    len-=32-offset;
+    pos+=32-offset;
+  }
+}
+ 
+void
+BitmaskImpl::setFieldImpl(Uint32 data[],
+			  unsigned pos, unsigned l, const Uint32 src[])
+{
+  Uint32 word;
+  Uint32 offset;
+  Uint32 mask;
+  int i=0,stored=0;
+  int len= l;
+
+  while(len>0)
+  {
+    ndbout_c("len: %d", len);
+    word = pos >> 5;
+    offset = pos & 31;
+    
+    if(offset+len > 32)
+      stored = 32-offset;
+    else
+      stored = len;
+    
+    mask = ((1 << stored) - 1) << (i+offset)%32;
+    data[word] = (data[word] & ~mask) | ((src[i/32] << (i+offset)%32) & mask);
+    
+    i+=stored;
+    len-=32-offset;
+    pos+=32-offset;
+  }
+}
+
+#else
+
+#define DEBUG 0
+#include <Vector.hpp>
+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;
+};
+
+void require(bool b)
+{
+  if(!b) abort();
+}
+
+void 
+do_test(int 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;
+  Vector<Uint32> tmp;
+  
+  ndbout_c("Testing bitmask of size %d", bitmask_size);
+  Uint32 zero = 0;
+  alloc_mask.fill(sz32, zero);
+  test_mask.fill(sz32, zero);
+  tmp.fill(sz32, zero);
+  
+  for(int i = 0; i<1000; i++)
+  {
+    int pos = rand() % (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;
+	}
+      }
+      if(DEBUG)
+	ndbout_c("freeing [ %d %d ]", min, max);
+
+      require(pos >= min && pos < max);
+      BitmaskImpl::getField(sz32, test_mask.getBase(), min, max-min, 
+			    tmp.getBase());
+      if(memcmp(tmp.getBase(), alloc_list[j].data.getBase(),
+		((max - min) + 31) >> 5) != 0)
+      {
+	printf("mask: ");
+	size_t k;
+	Alloc& a = alloc_list[j];
+	for(k = 0; k<a.data.size(); k++)
+	  printf("%.8x ", a.data[k]);
+	printf("\n");
+
+	printf("field: ");
+	for(k = 0; k<(((max - min)+31)>>5); k++)
+	  printf("%.8x ", tmp.getBase()[k]);
+	printf("\n");
+	abort();
+      }
+      while(min < max)
+	BitmaskImpl::clear(sz32, alloc_mask.getBase(), min++);
+      alloc_list.erase(j);
+    }
+    else
+    {
+      // Bit was free
+      // 1) Check how much space is avaiable
+      // 2) Create new allocation of random size
+      // 3) Fill data with random data
+      // 4) Update alloc mask
+      while(pos+free < bitmask_size && 
+	    !BitmaskImpl::get(sz32, alloc_mask.getBase(), pos+free))
+	free++;
+
+      Uint32 sz = (rand() % free); sz = sz ? sz : 1;
+      Alloc a;
+      a.pos = pos;
+      a.size = sz;
+      a.data.fill(sz >> 5, zero);
+      if(DEBUG)
+	ndbout_c("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((rand() % 1000) > 500)
+	  BitmaskImpl::set((sz + 31) >> 5, a.data.getBase(), j);
+      }
+      if(DEBUG)
+      {
+	printf("mask: ");
+	size_t k;
+	for(k = 0; k<a.data.size(); k++)
+	  printf("%.8x ", a.data[k]);
+	printf("\n");
+      }
+      BitmaskImpl::setField(sz32, test_mask.getBase(), pos, sz, 
+			    a.data.getBase());
+      alloc_list.push_back(a);
+    }
+  }
+}
+
+template class Vector<Alloc>;
+template class Vector<Uint32>;
+
+#endif

ndb/src/common/util/Makefile.am

@@ -9,7 +9,24 @@ libgeneral_la_SOURCES = \
 	    NdbSqlUtil.cpp new.cpp \
             uucode.c random.c version.c \
             strdup.c \
-            ConfigValues.cpp ndb_init.c basestring_vsnprintf.c
+            ConfigValues.cpp ndb_init.c basestring_vsnprintf.c \
+            Bitmask.cpp
+
+EXTRA_PROGRAMS = testBitmask
+testBitmask_SOURCES =  testBitmask.cpp
+testBitmask_LDFLAGS = @ndb_bin_am_ldflags@ \
+  $(top_builddir)/ndb/src/libndbclient.la \
+         $(top_builddir)/dbug/libdbug.a \
+         $(top_builddir)/mysys/libmysys.a \
+         $(top_builddir)/strings/libmystrings.a
+
+testBitmask.cpp : Bitmask.cpp
+	rm -f testBitmask.cpp
+	@LN_CP_F@ Bitmask.cpp testBitmask.cpp
+
+testBitmask.o:	$(testBitmask_SOURCES)
+		$(CXXCOMPILE) -c $(INCLUDES) -D__TEST_BITMASK__ $<
+
 
 include $(top_srcdir)/ndb/config/common.mk.am
 include $(top_srcdir)/ndb/config/type_util.mk.am

ndb/src/kernel/blocks/dbdict/Dbdict.cpp

@@ -4843,7 +4843,16 @@ void Dbdict::handleTabInfo(SimpleProperties::Reader & it,
     nullCount += attrDesc.AttributeNullableFlag;
     
     const Uint32 aSz = (1 << attrDesc.AttributeSize);
-    const Uint32 sz = ((aSz * attrDesc.AttributeArraySize) + 31) >> 5;
+    Uint32 sz;
+    if(aSz != 1)
+    {
+      sz = ((aSz * attrDesc.AttributeArraySize) + 31) >> 5;
+    }    
+    else
+    {
+      sz = 0;
+      nullCount += attrDesc.AttributeArraySize;      
+    }
 
     recordLength += sz;
     if(attrDesc.AttributeKeyFlag){

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

@@ -500,7 +500,8 @@ struct Fragoperrec {
   Uint32 tableidFrag;
   Uint32 fragPointer;
   Uint32 attributeCount;
-  Uint32 freeNullBit;
+  Uint32 currNullBit;
+  Uint32 noOfNullBits;
   Uint32 noOfNewAttrCount;
   Uint32 charsetIndex;
   BlockReference lqhBlockrefFrag;
@@ -1630,6 +1631,12 @@ private:
                              Uint32  attrDescriptor,
                              Uint32  attrDes2);
 
+
+  bool readBitsNULLable(Uint32* outBuffer, AttributeHeader*, Uint32, Uint32);
+  bool updateBitsNULLable(Uint32* inBuffer, Uint32, Uint32);
+  bool readBitsNotNULL(Uint32* outBuffer, AttributeHeader*, Uint32, Uint32);
+  bool updateBitsNotNULL(Uint32* inBuffer, Uint32, Uint32);
+
 // *****************************************************************
 // Read char routines optionally (tXfrmFlag) apply strxfrm
 // *****************************************************************

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

@@ -96,9 +96,10 @@ void Dbtup::execTUPFRAGREQ(Signal* signal)
   fragOperPtr.p->fragidFrag = fragId;
   fragOperPtr.p->tableidFrag = regTabPtr.i;
   fragOperPtr.p->attributeCount = noOfAttributes;
-  fragOperPtr.p->freeNullBit = noOfNullAttr;
+  fragOperPtr.p->noOfNullBits = noOfNullAttr;
   fragOperPtr.p->noOfNewAttrCount = noOfNewAttr;
   fragOperPtr.p->charsetIndex = 0;
+  fragOperPtr.p->currNullBit = 0;
 
   ndbrequire(reqinfo == ZADDFRAG);
 
@@ -309,13 +310,12 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
     Uint32 firstTabDesIndex = regTabPtr.p->tabDescriptor + (attrId * ZAD_SIZE);
     setTabDescrWord(firstTabDesIndex, attrDescriptor);
     Uint32 attrLen = AttributeDescriptor::getSize(attrDescriptor);
-    Uint32 nullBitPos = 0;                  /* Default pos for NOT NULL attributes */
+    Uint32 nullBitPos = fragOperPtr.p->currNullBit;
+
     if (AttributeDescriptor::getNullable(attrDescriptor)) {
       if (!AttributeDescriptor::getDynamic(attrDescriptor)) {
         ljam();                                      /* NULL ATTR */
-        fragOperPtr.p->freeNullBit--;	                           /* STORE NULL BIT POSTITION */
-        nullBitPos = fragOperPtr.p->freeNullBit;
-        ndbrequire(fragOperPtr.p->freeNullBit < ZNIL);               /* Check not below zero */
+        fragOperPtr.p->currNullBit++;
       }//if
     } else {
       ljam();
@@ -331,10 +331,21 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
       case 2:
       {
         ljam();
-        Uint32 bitsUsed = AttributeDescriptor::getArraySize(attrDescriptor) * (1 << attrLen);
+	if(attrLen != 0)
+	{
+	  ljam();
+	  Uint32 bitsUsed = 
+	    AttributeDescriptor::getArraySize(attrDescriptor) * (1 << attrLen);
 	  regTabPtr.p->tupheadsize += ((bitsUsed + 31) >> 5);
 	  break;
 	}
+	else
+	{
+	  ljam();
+	  Uint32 bitCount = AttributeDescriptor::getArraySize(attrDescriptor);
+	  fragOperPtr.p->currNullBit += bitCount;
+	}
+      }
       default:
         ndbrequire(false);
         break;
@@ -375,7 +386,9 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
       addattrrefuseLab(signal, regFragPtr, fragOperPtr, regTabPtr.p, fragId);
       return;
     }//if
-    if (lastAttr && (fragOperPtr.p->freeNullBit != 0)) {
+    if (lastAttr && 
+	(fragOperPtr.p->currNullBit != fragOperPtr.p->noOfNullBits))
+    {
       ljam();
       terrorCode = ZINCONSISTENT_NULL_ATTRIBUTE_COUNT;
       addattrrefuseLab(signal, regFragPtr, fragOperPtr, regTabPtr.p, fragId);

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

@@ -40,7 +40,11 @@ Dbtup::setUpQueryRoutines(Tablerec* const regTabPtr)
       if ((AttributeDescriptor::getArrayType(attrDescriptor) == ZNON_ARRAY) ||
           (AttributeDescriptor::getArrayType(attrDescriptor) == ZFIXED_ARRAY)) {
         if (!AttributeDescriptor::getNullable(attrDescriptor)) {
-          if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 1) {
+	  if (AttributeDescriptor::getSize(attrDescriptor) == 0){
+	    ljam();
+            regTabPtr->readFunctionArray[i] = &Dbtup::readBitsNotNULL;
+            regTabPtr->updateFunctionArray[i] = &Dbtup::updateBitsNotNULL;
+	  } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 1){
             ljam();
             regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHOneWordNotNULL;
             regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHOneWordNotNULL;
@@ -61,7 +65,11 @@ Dbtup::setUpQueryRoutines(Tablerec* const regTabPtr)
             regTabPtr->readFunctionArray[i] = &Dbtup::readCharNotNULL;
           }
         } else {
-          if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 1) {
+	  if (AttributeDescriptor::getSize(attrDescriptor) == 0){
+            ljam();
+            regTabPtr->readFunctionArray[i] = &Dbtup::readBitsNULLable;
+            regTabPtr->updateFunctionArray[i] = &Dbtup::updateBitsNULLable;
+	  } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 1){
             ljam();
             regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHOneWordNULLable;
             regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNULLable;
@@ -1005,3 +1013,178 @@ Dbtup::read_psuedo(Uint32 attrId, Uint32* outBuffer){
     return 0;
   }
 }
+
+bool
+Dbtup::readBitsNotNULL(Uint32* outBuffer,
+		       AttributeHeader* ahOut,
+		       Uint32  attrDescriptor,
+		       Uint32  attrDes2)
+{
+  Tablerec* const regTabPtr = tabptr.p;
+  Uint32 bitCount = AttributeDescriptor::getArraySize(attrDescriptor);
+  Uint32 offsetInTuple = AttributeOffset::getNullFlagOffset(attrDes2);
+  Uint32 offsetInWord = AttributeOffset::getNullFlagBitOffset(attrDes2);
+  ndbrequire(offsetInTuple < regTabPtr->tupNullWords);
+  offsetInTuple += regTabPtr->tupNullIndex;
+  ndbrequire(offsetInTuple < tCheckOffset);
+
+  Uint32 pos = offsetInTuple << 5 + offsetInWord;
+  Uint32 indexBuf = tOutBufIndex;
+  Uint32 newIndexBuf = indexBuf + ((bitCount + 31) >> 5);
+  Uint32 maxRead = tMaxRead;
+  
+  if (newIndexBuf <= maxRead) {
+    ljam();
+    ahOut->setDataSize((bitCount + 31) >> 5);
+    tOutBufIndex = newIndexBuf;
+    
+    BitmaskImpl::getField(regTabPtr->tupNullWords,
+			  tTupleHeader+regTabPtr->tupNullIndex,
+			  pos, 
+			  bitCount,
+			  outBuffer+indexBuf);
+    
+    return true;
+  } else {
+    ljam();
+    terrorCode = ZTRY_TO_READ_TOO_MUCH_ERROR;
+    return false;
+  }//if
+}
+
+bool
+Dbtup::readBitsNULLable(Uint32* outBuffer,
+			AttributeHeader* ahOut,
+			Uint32  attrDescriptor,
+			Uint32  attrDes2)
+{
+  Tablerec* const regTabPtr = tabptr.p;
+  Uint32 bitCount = AttributeDescriptor::getArraySize(attrDescriptor);
+  Uint32 offsetInTuple = AttributeOffset::getNullFlagOffset(attrDes2);
+  Uint32 offsetInWord = AttributeOffset::getNullFlagBitOffset(attrDes2);
+  ndbrequire(offsetInTuple < regTabPtr->tupNullWords);
+  offsetInTuple += regTabPtr->tupNullIndex;
+  ndbrequire(offsetInTuple < tCheckOffset);
+
+  Uint32 indexBuf = tOutBufIndex;
+  Uint32 newIndexBuf = indexBuf + (bitCount + 31) >> 5;
+  Uint32 maxRead = tMaxRead;
+  Uint32 pos = offsetInWord << 5 + offsetInTuple;
+  
+  if(BitmaskImpl::get(regTabPtr->tupNullWords,
+		      tTupleHeader+regTabPtr->tupNullIndex,
+		      pos))
+  {
+    ljam();
+    ahOut->setNULL();
+    return true;
+  }
+
+
+  if (newIndexBuf <= maxRead) {
+    ljam();
+    ahOut->setDataSize((bitCount + 31) >> 5);
+    tOutBufIndex = newIndexBuf;
+    BitmaskImpl::getField(regTabPtr->tupNullWords,
+			  tTupleHeader+regTabPtr->tupNullIndex,
+			  pos+1, 
+			  bitCount,
+			  outBuffer+indexBuf);
+    return true;
+  } else {
+    ljam();
+    terrorCode = ZTRY_TO_READ_TOO_MUCH_ERROR;
+    return false;
+  }//if
+}
+
+bool
+Dbtup::updateBitsNotNULL(Uint32* inBuffer,
+			 Uint32  attrDescriptor,
+			 Uint32  attrDes2)
+{
+  Tablerec* const regTabPtr =  tabptr.p;
+  Uint32 indexBuf = tInBufIndex;
+  Uint32 inBufLen = tInBufLen;
+  AttributeHeader ahIn(inBuffer[indexBuf]);
+  Uint32 nullIndicator = ahIn.isNULL();
+  Uint32 bitCount = AttributeDescriptor::getArraySize(attrDescriptor);
+  Uint32 newIndex = indexBuf + 1 + ((bitCount + 31) >> 5);
+  Uint32 nullFlagOffset = AttributeOffset::getNullFlagOffset(attrDes2);
+  Uint32 nullFlagBitOffset = AttributeOffset::getNullFlagBitOffset(attrDes2);
+  Uint32 nullWordOffset = nullFlagOffset + regTabPtr->tupNullIndex;
+  ndbrequire((nullFlagOffset < regTabPtr->tupNullWords) &&
+             (nullWordOffset < tCheckOffset));
+  Uint32 nullBits = tTupleHeader[nullWordOffset];
+  Uint32 pos = (nullFlagOffset << 5) + nullFlagBitOffset;
+  
+  if (newIndex <= inBufLen) {
+    if (!nullIndicator) {
+      BitmaskImpl::setField(regTabPtr->tupNullWords,
+			    tTupleHeader+regTabPtr->tupNullIndex,
+			    pos,
+			    bitCount,
+			    inBuffer+indexBuf+1);
+      tInBufIndex = newIndex;
+      return true;
+    } else {
+      ljam();
+      terrorCode = ZNOT_NULL_ATTR;
+      return false;
+    }//if
+  } else {
+    ljam();
+    terrorCode = ZAI_INCONSISTENCY_ERROR;
+    return false;
+  }//if
+  return true;
+}
+
+bool
+Dbtup::updateBitsNULLable(Uint32* inBuffer,
+			  Uint32  attrDescriptor,
+			  Uint32  attrDes2)
+{
+  Tablerec* const regTabPtr =  tabptr.p;
+  AttributeHeader ahIn(inBuffer[tInBufIndex]);
+  Uint32 indexBuf = tInBufIndex;
+  Uint32 nullIndicator = ahIn.isNULL();
+  Uint32 nullFlagOffset = AttributeOffset::getNullFlagOffset(attrDes2);
+  Uint32 nullFlagBitOffset = AttributeOffset::getNullFlagBitOffset(attrDes2);
+  Uint32 nullWordOffset = nullFlagOffset + regTabPtr->tupNullIndex;
+  ndbrequire((nullFlagOffset < regTabPtr->tupNullWords) &&
+             (nullWordOffset < tCheckOffset));
+  Uint32 nullBits = tTupleHeader[nullWordOffset];
+  Uint32 bitCount = AttributeDescriptor::getArraySize(attrDescriptor);  
+  Uint32 pos = (nullFlagOffset << 5) + nullFlagBitOffset;
+
+  if (!nullIndicator) {
+    BitmaskImpl::clear(regTabPtr->tupNullWords,
+		       tTupleHeader+regTabPtr->tupNullIndex,
+		       pos);
+    BitmaskImpl::setField(regTabPtr->tupNullWords,
+			  tTupleHeader+regTabPtr->tupNullIndex,
+			  pos+1,
+			  bitCount,
+			  inBuffer+indexBuf+1);
+
+    Uint32 newIndex = indexBuf + 1 + ((bitCount + 31) >> 5);
+    tInBufLen = newIndex;
+    return true;
+  } else {
+    Uint32 newIndex = tInBufIndex + 1;
+    if (newIndex <= tInBufLen) {
+      ljam();
+      BitmaskImpl::set(regTabPtr->tupNullWords,
+		       tTupleHeader+regTabPtr->tupNullIndex,
+		       pos);
+      
+      tInBufIndex = newIndex;
+      return true;
+    } else {
+      ljam();
+      terrorCode = ZAI_INCONSISTENCY_ERROR;
+      return false;
+    }//if
+  }//if
+}

ndb/test/ndbapi/Makefile.am

@@ -31,7 +31,8 @@ testTimeout \
 testTransactions \
 testDeadlock \
 test_event ndbapi_slow_select testReadPerf testLcp \
-testPartitioning
+testPartitioning \
+testBitfield
 
 #flexTimedAsynch
 #testBlobs
@@ -71,6 +72,7 @@ ndbapi_slow_select_SOURCES = slow_select.cpp
 testReadPerf_SOURCES = testReadPerf.cpp
 testLcp_SOURCES = testLcp.cpp
 testPartitioning_SOURCES = testPartitioning.cpp
+testBitfield_SOURCES = testBitfield.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/ndb/include/kernel
 

ndb/test/ndbapi/testBitfield.cpp

+
+#include <ndb_global.h>
+#include <ndb_opts.h>
+#include <NDBT.hpp>
+#include <NdbApi.hpp>
+
+static const char* opt_connect_str= 0;
+static const char* _dbname = "TEST_DB";
+static int g_loops = 5;
+static struct my_option my_long_options[] =
+{
+  NDB_STD_OPTS("ndb_desc"),
+  { "database", 'd', "Name of database table is in",
+    (gptr*) &_dbname, (gptr*) &_dbname, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
+};
+
+static void print_version()
+{
+  printf("MySQL distrib %s, for %s (%s)\n",
+	 MYSQL_SERVER_VERSION,SYSTEM_TYPE,MACHINE_TYPE);
+}
+static void usage()
+{
+  char desc[] = 
+    "tabname\n"\
+    "This program list all properties of table(s) in NDB Cluster.\n"\
+    "  ex: desc T1 T2 T4\n";  
+  print_version();
+  my_print_help(my_long_options);
+  my_print_variables(my_long_options);
+}
+static my_bool
+get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
+	       char *argument)
+{
+  switch (optid) {
+  case '#':
+    DBUG_PUSH(argument ? argument : "d:t:O,/tmp/ndb_desc.trace");
+    break;
+  case 'V':
+    print_version();
+    exit(0);
+  case '?':
+    usage();
+    exit(0);
+  }
+  return 0;
+}
+
+static const NdbDictionary::Table* create_random_table(Ndb*);
+static int transactions(Ndb*, const NdbDictionary::Table* tab);
+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);
+
+int 
+main(int argc, char** argv){
+  NDB_INIT(argv[0]);
+  const char *load_default_groups[]= { "mysql_cluster",0 };
+  load_defaults("my",load_default_groups,&argc,&argv);
+  int ho_error;
+  if ((ho_error=handle_options(&argc, &argv, my_long_options, get_one_option)))
+    return NDBT_ProgramExit(NDBT_WRONGARGS);
+
+  Ndb::setConnectString(opt_connect_str);
+
+  Ndb* pNdb;
+  pNdb = new Ndb(_dbname);  
+  pNdb->init();
+  while (pNdb->waitUntilReady() != 0);
+  int res = NDBT_FAILED;
+
+  NdbDictionary::Dictionary * dict = pNdb->getDictionary();
+
+  const NdbDictionary::Table* pTab = 0;
+  for (int i = 0; i < (argc ? argc : g_loops) ; i++)
+  {
+    res = NDBT_FAILED;
+    if(argc == 0)
+    {
+      pTab = create_random_table(pNdb);
+    }
+    else
+    {
+      dict->dropTable(argv[i]);
+      NDBT_Tables::createTable(pNdb, argv[i]);
+      pTab = dict->getTable(argv[i]);
+    }
+    
+    if (pTab == 0)
+    {
+      ndbout << "Failed to create table" << endl;
+      ndbout << dict->getNdbError() << endl;
+      break;
+    }
+    
+    if(transactions(pNdb, pTab))
+      break;
+
+    if(unique_indexes(pNdb, pTab))
+      break;
+
+    if(ordered_indexes(pNdb, pTab))
+      break;
+    
+    if(node_restart(pNdb, pTab))
+      break;
+    
+    if(system_restart(pNdb, pTab))
+      break;
+
+    dict->dropTable(pTab->getName());
+    res = NDBT_OK;
+  }
+
+  if(res != NDBT_OK && pTab)
+  {
+    dict->dropTable(pTab->getName());
+  }
+  
+  delete pNdb;
+  return NDBT_ProgramExit(res);
+}
+
+static 
+const NdbDictionary::Table* 
+create_random_table(Ndb* pNdb)
+{
+  return 0;
+}
+static 
+int
+transactions(Ndb* pNdb, const NdbDictionary::Table* tab)
+{
+  return 0;
+}
+
+static 
+int 
+unique_indexes(Ndb* pNdb, const NdbDictionary::Table* tab)
+{
+  return 0;
+}
+
+static 
+int 
+ordered_indexes(Ndb* pNdb, const NdbDictionary::Table* tab)
+{
+  return 0;
+}
+
+static 
+int 
+node_restart(Ndb* pNdb, const NdbDictionary::Table* tab)
+{
+  return 0;
+}
+
+static 
+int 
+system_restart(Ndb* pNdb, const NdbDictionary::Table* tab)
+{
+  return 0;
+}