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

into mysql.com:/home/jonas/src/wl2126


ndb/include/ndbapi/NdbDictionary.hpp:
  Auto merged
ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Auto merged
ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp:
  Auto merged

4 files changed, 217 insertions, 13 deletions

diff --git a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
index 4c34d2ce0db..1c7662c5d55 100644
--- a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
+++ b/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
 // *****************************************************************
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp b/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
index 49de0d80bcd..8e3ca6528c2 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
+++ b/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
@@ -1113,6 +1113,7 @@ Dbtup::updateStartLab(Signal* signal,
                                 regOperPtr->attrinbufLen);
     if (retValue == -1) {
       tupkeyErrorLab(signal);
+      return -1;
     }//if
   } else {
     jam();
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp b/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
index 405f790954e..fc27cca9cda 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
+++ b/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 */
+        ljam();                                      /* NULL ATTR */
+        fragOperPtr.p->currNullBit++;
       }//if
     } else {
       ljam();
@@ -331,9 +331,20 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
       case 2:
       {
         ljam();
-        Uint32 bitsUsed = AttributeDescriptor::getArraySize(attrDescriptor) * (1 << attrLen);
-        regTabPtr.p->tupheadsize += ((bitsUsed + 31) >> 5);
-        break;
+	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);
@@ -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);
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp b/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
index 8c1ba24d710..60da2797f60 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
+++ b/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;
@@ -1012,3 +1020,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
+}