wl2240 - fix handling of distribution key wrt unique indexes

ndb/include/kernel/AttributeDescriptor.hpp:
  removed unused
ndb/include/kernel/ndb_limits.h:
  removed unused
ndb/src/kernel/blocks/dbdict/Dbdict.cpp:
  Differentiate between nullable columns and null fields
    TUP wants sum all other want #null cols
ndb/src/kernel/blocks/dbdict/Dbdict.hpp:
  Differentiate between nullable columns and null fields
    TUP wants sum all other want #null cols
ndb/src/kernel/blocks/dbtc/Dbtc.hpp:
  Handle distribution key and collation in same subroutine
ndb/src/kernel/blocks/dbtc/DbtcMain.cpp:
  Handle distribution key and collation in same subroutine
ndb/src/kernel/vm/VMSignal.hpp:
  Increase size of signal->theData to 32k
ndb/test/ndbapi/testPartitioning.cpp:
  better usage

ndb/include/kernel/AttributeDescriptor.hpp

@@ -30,7 +30,6 @@ private:
   static void setArray(Uint32 &, Uint32 arraySize);
   static void setOriginal(Uint32 &, Uint32 original);
   static void setNullable(Uint32 &, Uint32 nullable);
-  static void setDGroup(Uint32 &, Uint32 dgroup);
   static void setDKey(Uint32 &, Uint32 dkey);
   static void setPrimaryKey(Uint32 &, Uint32 dkey);
   static void setDynamic(Uint32 &, Uint32 dynamicInd);

ndb/include/kernel/ndb_limits.h

@@ -50,7 +50,6 @@
  **/
 #define MAX_TUPLES_PER_PAGE 8191
 #define MAX_TUPLES_BITS 13 		/* 13 bits = 8191 tuples per page */
-/*#define MAX_NO_OF_TUPLEKEY 16 Not currently used */
 #define MAX_TABLES 1600
 #define MAX_TAB_NAME_SIZE 128
 #define MAX_ATTR_NAME_SIZE 32
@@ -58,7 +57,6 @@
 #define MAX_ATTRIBUTES_IN_TABLE 128
 #define MAX_ATTRIBUTES_IN_INDEX 32
 #define MAX_TUPLE_SIZE_IN_WORDS 2013
-#define MAX_FIXED_KEY_LENGTH_IN_WORDS 8
 #define MAX_KEY_SIZE_IN_WORDS 1023
 #define MAX_FRM_DATA_SIZE 6000
 #define MAX_NULL_BITS 4096

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

@@ -4098,7 +4098,7 @@ Dbdict::execADD_FRAGREQ(Signal* signal) {
     req->lh3DistrBits = 0; //lhDistrBits;
     req->lh3PageBits = 0; //lhPageBits;
     req->noOfAttributes = tabPtr.p->noOfAttributes;
-    req->noOfNullAttributes = tabPtr.p->noOfNullAttr;
+    req->noOfNullAttributes = tabPtr.p->noOfNullBits;
     req->noOfPagesToPreAllocate = 0;
     req->schemaVersion = tabPtr.p->tableVersion;
     Uint32 keyLen = tabPtr.p->tupKeyLength;
@@ -4739,6 +4739,7 @@ void Dbdict::handleTabInfo(SimpleProperties::Reader & it,
   Uint32 keyLength = 0;
   Uint32 attrCount = tablePtr.p->noOfAttributes;
   Uint32 nullCount = 0;
+  Uint32 nullBits = 0;
   Uint32 noOfCharsets = 0;
   Uint16 charsets[128];
   Uint32 recordLength = 0;
@@ -4867,9 +4868,9 @@ void Dbdict::handleTabInfo(SimpleProperties::Reader & it,
     else
     {
       sz = 0;
-      nullCount += attrDesc.AttributeArraySize;      
+      nullBits += attrDesc.AttributeArraySize;      
     }
-
+    
     recordLength += sz;
     if(attrDesc.AttributeKeyFlag){
       keyLength += sz;
@@ -4897,6 +4898,7 @@ void Dbdict::handleTabInfo(SimpleProperties::Reader & it,
   tablePtr.p->noOfNullAttr = nullCount;
   tablePtr.p->noOfCharsets = noOfCharsets;
   tablePtr.p->tupKeyLength = keyLength;
+  tablePtr.p->noOfNullBits = nullCount + nullBits;
 
   tabRequire(recordLength<= MAX_TUPLE_SIZE_IN_WORDS, 
 	     CreateTableRef::RecordTooBig);

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

@@ -213,7 +213,9 @@ public:
       IL_CREATED_TC = 1 << 0    // created in TC
     };
     Uint32 indexLocal;
-
+    
+    Uint32 noOfNullBits;
+    
     inline bool equal(TableRecord & rec) const {
       return strcmp(tableName, rec.tableName) == 0;
     }

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

@@ -897,7 +897,7 @@ public:
     UintR  hashValue;    /* THE HASH VALUE USED TO LOCATE FRAGMENT       */
     
     Uint8  distributionKeyIndicator;
-    Uint8  unused1;
+    Uint8  m_special_hash; // collation or distribution key
     Uint8  unused2;
     Uint8  lenAiInTckeyreq;  /* LENGTH OF ATTRIBUTE INFORMATION IN TCKEYREQ */
 
@@ -986,7 +986,8 @@ public:
 
     Uint8 noOfKeyAttr;
     Uint8 hasCharAttr;
-
+    Uint8 noOfDistrKeys;
+    
     struct KeyAttr {
       Uint32 attributeDescriptor;
       CHARSET_INFO* charsetInfo;
@@ -1445,7 +1446,10 @@ private:
   void gcpTcfinished(Signal* signal);
   void handleGcp(Signal* signal);
   void hash(Signal* signal);
-  Uint32 xfrmKeyData(Signal* signal, Uint32* dst, Uint32 dstSize, const Uint32* src);
+  Uint32 handle_special_hash(Uint32 dstHash[4], 
+			     Uint32* src, Uint32 srcLen, 
+			     Uint32 tabPtrI, bool distr);
+  
   void initApiConnect(Signal* signal);
   void initApiConnectRec(Signal* signal, 
 			 ApiConnectRecord * const regApiPtr,

ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -330,7 +330,7 @@ void Dbtc::execTC_SCHVERREQ(Signal* signal)
   Uint32 noOfKeyAttr = signal->theData[6];
   ndbrequire(noOfKeyAttr <= MAX_ATTRIBUTES_IN_INDEX);
   Uint32 hasCharAttr = 0;
-
+  Uint32 noOfDistrKeys = 0;
   SegmentedSectionPtr s0Ptr;
   signal->getSection(s0Ptr, 0);
   SectionReader r0(s0Ptr, getSectionSegmentPool());
@@ -350,6 +350,8 @@ void Dbtc::execTC_SCHVERREQ(Signal* signal)
       ndbrequire(cs != 0);
       hasCharAttr = 1;
     }
+    
+    noOfDistrKeys += AttributeDescriptor::getDKey(attributeDescriptor);
     tabptr.p->keyAttr[i].attributeDescriptor = attributeDescriptor;
     tabptr.p->keyAttr[i].charsetInfo = cs;
     i++;
@@ -362,6 +364,7 @@ void Dbtc::execTC_SCHVERREQ(Signal* signal)
   tabptr.p->dropping = false;
   tabptr.p->noOfKeyAttr = noOfKeyAttr;
   tabptr.p->hasCharAttr = hasCharAttr;
+  tabptr.p->noOfDistrKeys = noOfDistrKeys;
   
   signal->theData[0] = tabptr.i;
   signal->theData[1] = retPtr;
@@ -2258,12 +2261,10 @@ void Dbtc::hash(Signal* signal)
   UintR  Tdata2;
   UintR  Tdata3;
   UintR*  Tdata32;
-  Uint64 Tdata[512];
-  Uint64 Txfrmdata[512 * MAX_XFRM_MULTIPLY];
-
+  
   CacheRecord * const regCachePtr = cachePtr.p;
-  Tdata32 = (UintR*)&Tdata[0];
-
+  Tdata32 = signal->theData;
+  
   Tdata0 = regCachePtr->keydata[0];
   Tdata1 = regCachePtr->keydata[1];
   Tdata2 = regCachePtr->keydata[2];
@@ -2291,22 +2292,20 @@ void Dbtc::hash(Signal* signal)
   }//if
 
   UintR keylen = (UintR)regCachePtr->keylen;
-
-  TableRecordPtr tabptrSave = tabptr;
-  tabptr.i = regCachePtr->tableref; // table or hash index id
-  ptrCheckGuard(tabptr, ctabrecFilesize, tableRecord);
-  if (tabptr.p->hasCharAttr) {
-    jam();
-    keylen = xfrmKeyData(signal, (Uint32*)Txfrmdata, sizeof(Txfrmdata) >> 2, Tdata32);
-    Tdata32 = (UintR*)&Txfrmdata[0];
-  }
-  tabptr = tabptrSave;
-
+  Uint32 distKey = regCachePtr->distributionKeyIndicator;
+  
   Uint32 tmp[4];
-  md5_hash(tmp, (Uint64*)&Tdata32[0], keylen);
-
+  if(!regCachePtr->m_special_hash)
+  {
+    md5_hash(tmp, (Uint64*)&Tdata32[0], keylen);
+  }
+  else
+  {
+    handle_special_hash(tmp, Tdata32, keylen, regCachePtr->tableref, !distKey);
+  }
+  
   thashValue = tmp[0];
-  if (regCachePtr->distributionKeyIndicator == 1) {
+  if (distKey){
     jam();
     tdistrHashValue = regCachePtr->distributionKey;
   } else {
@@ -2316,44 +2315,109 @@ void Dbtc::hash(Signal* signal)
 }//Dbtc::hash()
 
 Uint32
-Dbtc::xfrmKeyData(Signal* signal, Uint32* dst, Uint32 dstSize, const Uint32* src)
-{
-  const Uint32 noOfKeyAttr = tabptr.p->noOfKeyAttr;
+Dbtc::handle_special_hash(Uint32 dstHash[4], Uint32* src, Uint32 srcLen, 
+			  Uint32 tabPtrI,
+			  bool distr)
+{
+  Uint64 Tmp[MAX_KEY_SIZE_IN_WORDS * 4 * MAX_XFRM_MULTIPLY];
+  const Uint32 dstSize = sizeof(Tmp) / 4;
+  const TableRecord* tabPtrP = &tableRecord[tabPtrI];
+  const Uint32 noOfKeyAttr = tabPtrP->noOfKeyAttr;
+  Uint32 noOfDistrKeys = tabPtrP->noOfDistrKeys;
+  const bool hasCharAttr = tabPtrP->hasCharAttr;
+  
+  Uint32 *dst = (Uint32*)Tmp;
   Uint32 dstPos = 0;
   Uint32 srcPos = 0;
-  Uint32 i = 0;
-  while (i < noOfKeyAttr) {
-    const TableRecord::KeyAttr& keyAttr = tabptr.p->keyAttr[i];
-
-    Uint32 srcBytes = AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
-    Uint32 srcWords = (srcBytes + 3) / 4;
-    Uint32 dstWords = ~0;
-    uchar* dstPtr = (uchar*)&dst[dstPos];
-    const uchar* srcPtr = (const uchar*)&src[srcPos];
-    CHARSET_INFO* cs = keyAttr.charsetInfo;
-
-    if (cs == NULL) {
-      jam();
-      memcpy(dstPtr, srcPtr, srcWords << 2);
-      dstWords = srcWords;
-    } else {
-      jam();
-      Uint32 xmul = cs->strxfrm_multiply;
-      if (xmul == 0)
-        xmul = 1;
-      Uint32 dstLen = xmul * srcBytes;
-      ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
-      uint n = (*cs->coll->strnxfrm)(cs, dstPtr, dstLen, srcPtr, srcBytes);
-      while ((n & 3) != 0) {
-        dstPtr[n++] = 0;
+  Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
+  if(hasCharAttr){
+    Uint32 i = 0;
+    while (i < noOfKeyAttr) {
+      const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
+      
+      Uint32 srcBytes = 
+	AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
+      Uint32 srcWords = (srcBytes + 3) / 4;
+      Uint32 dstWords = ~0;
+      uchar* dstPtr = (uchar*)&dst[dstPos];
+      const uchar* srcPtr = (const uchar*)&src[srcPos];
+      CHARSET_INFO* cs = keyAttr.charsetInfo;
+      
+      if (cs == NULL) {
+	jam();
+	memcpy(dstPtr, srcPtr, srcWords << 2);
+	dstWords = srcWords;
+      } else {
+	jam();
+	Uint32 xmul = cs->strxfrm_multiply;
+	if (xmul == 0)
+	  xmul = 1;
+	Uint32 dstLen = xmul * srcBytes;
+	ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
+	uint n = (*cs->coll->strnxfrm)(cs, dstPtr, dstLen, srcPtr, srcBytes);
+	while ((n & 3) != 0) {
+	  dstPtr[n++] = 0;
+	}
+	dstWords = (n >> 2);
+	
       }
-      dstWords = (n >> 2);
+      dstPos += dstWords;
+      srcPos += srcWords;
+      keyPartLen[i++] = dstWords;
     }
-    dstPos += dstWords;
-    srcPos += srcWords;
-    i++;
+  } 
+  else 
+  {
+    dst = src;
+    dstPos = srcLen;
+  }
+  
+  md5_hash(dstHash, (Uint64*)dst, dstPos);
+  
+  if(distr && noOfDistrKeys)
+  {
+    jam();
+    src = dst;
+    dstPos = 0;
+    Uint32 i = 0;
+    if(hasCharAttr)
+    {
+      while (i < noOfKeyAttr && noOfDistrKeys) 
+      {
+	const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
+	Uint32 len = keyPartLen[i];
+	if(AttributeDescriptor::getDKey(keyAttr.attributeDescriptor))
+	{
+	  noOfDistrKeys--;
+	  memmove(dst+dstPos, src, len << 2);
+	  dstPos += len;
+	}
+	src += len;
+	i++;
+      }
+    }
+    else
+    {
+      while (i < noOfKeyAttr && noOfDistrKeys) 
+      {
+	const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
+	Uint32 len = 
+	  AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
+	len = (len + 3) / 4;
+	if(AttributeDescriptor::getDKey(keyAttr.attributeDescriptor))
+	{
+	  noOfDistrKeys--;
+	  memmove(dst+dstPos, src, len << 2);
+	  dstPos += len;
+	}
+	src += len;
+	i++;
+      }
+    }
+    Uint32 tmp[4];
+    md5_hash(tmp, (Uint64*)dst, dstPos);
+    dstHash[1] = tmp[1];
   }
-  return dstPos;
 }
 
 /*
@@ -2803,7 +2867,8 @@ void Dbtc::execTCKEYREQ(Signal* signal)
   regCachePtr->keylen = TkeyLength;
   regCachePtr->lenAiInTckeyreq = titcLenAiInTckeyreq;
   regCachePtr->currReclenAi = titcLenAiInTckeyreq;
-
+  regCachePtr->m_special_hash = 
+    localTabptr.p->hasCharAttr | (localTabptr.p->noOfDistrKeys > 0);
   Tdata1 = TAIDataPtr[0];
   Tdata2 = TAIDataPtr[1];
   Tdata3 = TAIDataPtr[2];
@@ -10070,6 +10135,7 @@ void Dbtc::initTable(Signal* signal)
     tabptr.p->dropping = false;
     tabptr.p->noOfKeyAttr = 0;
     tabptr.p->hasCharAttr = 0;
+    tabptr.p->noOfDistrKeys = 0;
     for (unsigned k = 0; k < MAX_ATTRIBUTES_IN_INDEX; k++) {
       tabptr.p->keyAttr[k].attributeDescriptor = 0;
       tabptr.p->keyAttr[k].charsetInfo = 0;

ndb/src/kernel/vm/VMSignal.hpp

@@ -78,10 +78,16 @@ public:
 #define VMS_DATA_SIZE \
   (MAX_ATTRIBUTES_IN_TABLE + MAX_TUPLE_SIZE_IN_WORDS + MAX_KEY_SIZE_IN_WORDS)
 
+#if VMS_DATA_SIZE > 8192
+#error "VMSignal buffer is too small"
+#endif
+  
   SignalHeader header; // 28 bytes
   SegmentedSectionPtr m_sectionPtr[3]; 
-  Uint32 theData[25+VMS_DATA_SIZE];  // 2048 32-bit words -> 8K Bytes
-  
+  union {
+    Uint32 theData[8192];  // 8192 32-bit words -> 32K Bytes
+    Uint64 dummyAlign;
+  };
   void garbage_register();
 };
 

ndb/test/ndbapi/testPartitioning.cpp

@@ -84,7 +84,7 @@ add_distribution_key(Ndb*, NdbDictionary::Table& tab, int when)
       col.setType(NdbDictionary::Column::Unsigned);
       col.setLength(1); 
       col.setNullable(false);
-      col.setPrimaryKey(1);
+      col.setPrimaryKey(true);
       col.setDistributionKey(true);
       tab.addColumn(col);
     }