bzr commit into mysql-5.1 branch (tomas.ulin:3074)

[Tomas Ulin]

#At file:///home/tomas/mysql_src/mysql-5.1-telco-6.4/

 3074 Tomas Ulin	2008-11-13
      new start transaction with hint methods using ndb record
modified:
  storage/ndb/include/ndbapi/Ndb.hpp
  storage/ndb/src/ndbapi/Ndb.cpp

=== modified file 'storage/ndb/include/ndbapi/Ndb.hpp'
--- a/storage/ndb/include/ndbapi/Ndb.hpp	2008-11-06 16:52:59 +0000
+++ b/storage/ndb/include/ndbapi/Ndb.hpp	2008-11-13 15:30:02 +0000
@@ -1374,7 +1374,10 @@ public:
   NdbTransaction* startTransaction(const NdbDictionary::Table *table,
 				   const struct Key_part_ptr * keyData,
 				   void* xfrmbuf = 0, Uint32 xfrmbuflen = 0);
-
+#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
+  NdbTransaction* startTransaction(const NdbRecord *keyRec, const char *keyData,
+				   void* xfrmbuf, Uint32 xfrmbuflen);
+#endif
   /**
    * Start a transaction, specifying table+partition as hint for
    *  TC-selection
@@ -1406,7 +1409,11 @@ public:
                          const NdbDictionary::Table*, 
                          const struct Key_part_ptr * keyData,
                          void* xfrmbuf = 0, Uint32 xfrmbuflen = 0);
-  
+#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
+  static int computeHash(Uint32* hashvalueptr,
+                         const NdbRecord *keyRec, const char *keyData,
+                         void* xfrmbuf, Uint32 xfrmbuflen);
+#endif
   /**
    * Close a transaction.
    *

=== modified file 'storage/ndb/src/ndbapi/Ndb.cpp'
--- a/storage/ndb/src/ndbapi/Ndb.cpp	2008-11-06 17:34:29 +0000
+++ b/storage/ndb/src/ndbapi/Ndb.cpp	2008-11-13 15:30:02 +0000
@@ -345,6 +345,7 @@ Ndb::computeHash(Uint32 *retval,
       partcols[j++] = cols[i];
     }
   }
+  DBUG_ASSERT(j == parts);
 
   for (Uint32 i = 0; i<parts; i++)
   {
@@ -478,6 +479,134 @@ emalformedkey:
 
 enomem:
   return 4000;
+}
+
+int
+Ndb::computeHash(Uint32 *retval,
+                 const NdbRecord *keyRec,
+                 const char *keyData, 
+                 void* buf, Uint32 bufLen)
+{
+  Uint32 len;
+  char* pos = (char*)buf;
+
+  Uint32 parts = keyRec->distkey_index_length;
+
+  {
+    UintPtr org = UintPtr(buf);
+    UintPtr use = (org + 7) & ~(UintPtr)7;
+
+    buf = (void*)use;
+    bufLen -= (use - org);
+  }
+
+  for (Uint32 i = 0; i < parts; i++)
+  {
+    const struct NdbRecord::Attr &keyAttr =
+      keyRec->columns[keyRec->distkey_indexes[i]];
+
+    Uint32 len;
+    Uint32 maxlen = keyAttr.maxSize;
+    unsigned char *src= (unsigned char*)keyData + keyAttr.offset;
+
+    if (keyAttr.flags & NdbRecord::IsVar1ByteLen)
+    {
+      if (keyAttr.flags & NdbRecord::IsMysqldShrinkVarchar)
+      {
+        len = uint2korr(src);
+        src += 2;
+      }
+      else
+      {
+        len = *src;
+        src += 1;
+      }
+      maxlen -= 1;
+    }
+    else if (keyAttr.flags & NdbRecord::IsVar2ByteLen)
+    {
+      len = uint2korr(src);
+      src += 2;
+      maxlen -= 2;
+    }
+    else
+    {
+      len = maxlen;
+    }
+
+    const CHARSET_INFO* cs = keyAttr.charset_info;
+    if (cs)
+    {
+      Uint32 xmul = cs->strxfrm_multiply;
+      if (xmul == 0)
+        xmul = 1;
+      /*
+       * Varchar end-spaces are ignored in comparisons.  To get same hash
+       * we blank-pad to maximum length via strnxfrm.
+       */
+      Uint32 dstLen = xmul * maxlen;
+      int n = NdbSqlUtil::strnxfrm_bug7284((CHARSET_INFO*)cs,
+                                           (unsigned char*)pos,
+                                           dstLen, src,
+                                           len);
+      if (unlikely(n == -1))
+        goto emalformedstring;
+      len = n;
+    }
+    else
+    {
+      if (keyAttr.flags & NdbRecord::IsVar1ByteLen)
+      {
+        *pos= (unsigned char)len;
+        memcpy(pos+1, src, len);
+        len += 1;
+      }
+      else if (keyAttr.flags & NdbRecord::IsVar2ByteLen)
+      {
+        len += 2;
+        memcpy(pos, src-2, len);
+      }
+      else
+        memcpy(pos, src, len);
+    }
+    while (len & 3)
+    {
+      * (pos + len++) = 0;
+    }
+    pos += len;
+  }
+  len = UintPtr(pos) - UintPtr(buf);
+  assert((len & 3) == 0);
+
+  Uint32 values[4];
+  md5_hash(values, (const Uint64*)buf, len >> 2);
+  
+  if (retval)
+  {
+    * retval = values[1];
+  }
+  
+  if (bufLen == 0)
+    free(buf);
+
+  return 0;
+  
+emalformedstring:  
+  return 4279;
+}
+
+NdbTransaction*
+Ndb::startTransaction(const NdbRecord *keyRec, const char *keyData,
+                      void* xfrmbuf, Uint32 xfrmbuflen)
+{
+  int ret;
+  Uint32 hash;
+  if ((ret = computeHash(&hash, keyRec, keyData, xfrmbuf, xfrmbuflen)) == 0)
+  {
+    return startTransaction(keyRec->table, keyRec->table->getPartitionId(hash));
+  }
+  theError.code = ret;
+  return 0;
 }
 
 NdbTransaction*