Merge pnousiainen@bk-internal.mysql.com:/home/bk/mysql-4.1-ndb

into mysql.com:/orca/space/pekka/ndb/version/my41-blobs

ndb/include/util/NdbSqlUtil.hpp

@@ -18,7 +18,7 @@
 #define NDB_SQL_UTIL_HPP
 
 #include <string.h>
-#include <ndb_types.h>
+#include <ndb_global.h>
 #include <kernel/ndb_limits.h>
 
 class NdbSqlUtil {
@@ -131,6 +131,7 @@ private:
 inline int
 NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
 {
+  // XXX require size >= 1
   if (size > full)
     return CmpError;
   switch ((Type::Enum)typeId) {
@@ -192,10 +193,38 @@ NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full,
       }
       return CmpUnknown;
     }
-  case Type::Mediumint:         // XXX fix these
-    break;
+  case Type::Mediumint:
+    {
+      if (size >= 1) {
+        union { const Uint32* p; const unsigned char* v; } u1, u2;
+        u1.p = p1;
+        u2.p = p2;
+        Int32 v1 = sint3korr(u1.v);
+        Int32 v2 = sint3korr(u2.v);
+        if (v1 < v2)
+          return -1;
+        if (v1 > v2)
+          return +1;
+        return 0;
+      }
+      return CmpUnknown;
+    }
   case Type::Mediumunsigned:
-    break;
+    {
+      if (size >= 1) {
+        union { const Uint32* p; const unsigned char* v; } u1, u2;
+        u1.p = p1;
+        u2.p = p2;
+        Uint32 v1 = uint3korr(u1.v);
+        Uint32 v2 = uint3korr(u2.v);
+        if (v1 < v2)
+          return -1;
+        if (v1 > v2)
+          return +1;
+        return 0;
+      }
+      return CmpUnknown;
+    }
   case Type::Int:
     {
       if (size >= 1) {
@@ -287,6 +316,7 @@ NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full,
       return CmpUnknown;
     }
   case Type::Decimal:
+    // XXX not used by MySQL or NDB
     break;
   case Type::Char:
     {
@@ -317,10 +347,28 @@ NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full,
       }
       return CmpUnknown;
     }
-  case Type::Binary:            // XXX fix these
-    break;
+  case Type::Binary:
+    {
+      // compare byte wise
+      union { const Uint32* p; const char* v; } u1, u2;
+      u1.p = p1;
+      u2.p = p2;
+      int k = memcmp(u1.v, u2.v, size << 2);
+      return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
+    }
   case Type::Varbinary:
-    break;
+    {
+      // assume correctly padded and compare byte wise
+      if (size >= 1) {
+        union { const Uint32* p; const char* v; } u1, u2;
+        u1.p = p1;
+        u2.p = p2;
+        // length in first 2 bytes
+        int k = memcmp(u1.v + 2, u2.v + 2, (size << 2) - 2);
+        return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
+      }
+      return CmpUnknown;
+    }
   case Type::Datetime:
     {
       /*
@@ -331,19 +379,57 @@ NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full,
         u1.p = p1;
         u2.p = p2;
         // skip format check
-        int k = strncmp(u1.v, u2.v, 4);
+        int k = memcmp(u1.v, u2.v, 4);
         if (k != 0)
-          return k;
+          return k < 0 ? -1 : +1;
         if (size >= 2) {
-          return strncmp(u1.v + 4, u2.v + 4, 4);
+          k = memcmp(u1.v + 4, u2.v + 4, 4);
+          return k < 0 ? -1 : k > 0 ? +1 : 0;
         }
       }
       return CmpUnknown;
     }
-  case Type::Timespec:          // XXX fix this
-    break;
-  case Type::Blob:              // XXX fix
-    break;
+  case Type::Timespec:
+    {
+      /*
+       * Timespec is CC YY MM DD hh mm ss \0 NN NN NN NN
+       */
+      if (size >= 1) {
+        union { const Uint32* p; const char* v; } u1, u2;
+        u1.p = p1;
+        u2.p = p2;
+        // skip format check
+        int k = memcmp(u1.v, u2.v, 4);
+        if (k != 0)
+          return k < 0 ? -1 : +1;
+        if (size >= 2) {
+          k = memcmp(u1.v + 4, u2.v + 4, 4);
+          if (k != 0)
+            return k < 0 ? -1 : +1;
+          Uint32 n1 = *(const Uint32*)(u1.v + 8);
+          Uint32 n2 = *(const Uint32*)(u2.v + 8);
+          if (n1 < n2)
+            return -1;
+          if (n2 > n1)
+            return +1;
+          return 0;
+        }
+      }
+      return CmpUnknown;
+    }
+  case Type::Blob:
+    {
+      // skip blob head, the rest is binary
+      const unsigned skip = NDB_BLOB_HEAD_SIZE;
+      if (size >= skip + 1) {
+        union { const Uint32* p; const char* v; } u1, u2;
+        u1.p = p1 + skip;
+        u2.p = p2 + skip;
+        int k = memcmp(u1.v, u2.v, (size - 1) << 2);
+        return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
+      }
+      return CmpUnknown;
+    }
   case Type::Text:
     {
       // skip blob head, the rest is char
@@ -352,7 +438,8 @@ NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full,
         union { const Uint32* p; const char* v; } u1, u2;
         u1.p = p1 + skip;
         u2.p = p2 + skip;
-        // TODO
+        int k = memcmp(u1.v, u2.v, (size - 1) << 2);
+        return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
       }
       return CmpUnknown;
     }

ndb/src/common/util/NdbSqlUtil.cpp

@@ -98,11 +98,11 @@ NdbSqlUtil::m_typeList[] = {
   },
   {
     Type::Mediumint,
-    NULL // cmpMediumint
+    cmpMediumint
   },
   {
     Type::Mediumunsigned,
-    NULL // cmpMediumunsigned
+    cmpMediumunsigned
   },
   {
     Type::Int,
@@ -130,7 +130,7 @@ NdbSqlUtil::m_typeList[] = {
   },
   {
     Type::Decimal,
-    NULL // cmpDecimal
+    NULL  // cmpDecimal
   },
   {
     Type::Char,
@@ -142,11 +142,11 @@ NdbSqlUtil::m_typeList[] = {
   },
   {
     Type::Binary,
-    NULL // cmpBinary
+    cmpBinary
   },
   {
     Type::Varbinary,
-    NULL // cmpVarbinary
+    cmpVarbinary
   },
   {
     Type::Datetime,
@@ -154,11 +154,11 @@ NdbSqlUtil::m_typeList[] = {
   },
   {
     Type::Timespec,
-    NULL // cmpTimespec
+    cmpTimespec
   },
   {
     Type::Blob,
-    NULL // cmpDatetime
+    cmpBlob
   },
   {
     Type::Text,