ndb - fix old decimal type mysql vs ndb

mysql-test/r/ndb_index_ordered.result:
  fix old decimal type mysql vs ndb (re-commit 3)
mysql-test/t/ndb_index_ordered.test:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/include/kernel/signaldata/DictTabInfo.hpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/include/ndbapi/NdbDictionary.hpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/include/util/NdbSqlUtil.hpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/src/common/util/NdbSqlUtil.cpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/src/ndbapi/NdbDictionary.cpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/src/ndbapi/NdbRecAttr.cpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/test/include/NdbSchemaOp.hpp:
  fix old decimal type mysql vs ndb (re-commit 3)
ndb/tools/restore/consumer.cpp:
  fix old decimal type mysql vs ndb (re-commit 3)
sql/ha_ndbcluster.cc:
  fix old decimal type mysql vs ndb (re-commit 3)

9 files changed, 142 insertions, 54 deletions

diff --git a/ndb/include/kernel/signaldata/DictTabInfo.hpp b/ndb/include/kernel/signaldata/DictTabInfo.hpp
index ade6c22a5bd..a2f9fcc9799 100644
--- a/ndb/include/kernel/signaldata/DictTabInfo.hpp
+++ b/ndb/include/kernel/signaldata/DictTabInfo.hpp
@@ -302,7 +302,8 @@ public:
     ExtBigunsigned = NdbSqlUtil::Type::Bigunsigned,
     ExtFloat = NdbSqlUtil::Type::Float,
     ExtDouble = NdbSqlUtil::Type::Double,
-    ExtDecimal = NdbSqlUtil::Type::Decimal,
+    ExtOlddecimal = NdbSqlUtil::Type::Olddecimal,
+    ExtOlddecimalunsigned = NdbSqlUtil::Type::Olddecimalunsigned,
     ExtChar = NdbSqlUtil::Type::Char,
     ExtVarchar = NdbSqlUtil::Type::Varchar,
     ExtBinary = NdbSqlUtil::Type::Binary,
@@ -411,9 +412,20 @@ public:
         AttributeSize = DictTabInfo::a64Bit;
         AttributeArraySize = AttributeExtLength;
         return true;
-      case DictTabInfo::ExtDecimal:
-        // not yet implemented anywhere
-        break;
+      case DictTabInfo::ExtOlddecimal:
+        AttributeType = DictTabInfo::StringType;
+        AttributeSize = DictTabInfo::an8Bit;
+        AttributeArraySize =
+          (1 + AttributeExtPrecision + (int(AttributeExtScale) > 0)) *
+          AttributeExtLength;
+        return true;
+      case DictTabInfo::ExtOlddecimalunsigned:
+        AttributeType = DictTabInfo::StringType;
+        AttributeSize = DictTabInfo::an8Bit;
+        AttributeArraySize =
+          (0 + AttributeExtPrecision + (int(AttributeExtScale) > 0)) *
+          AttributeExtLength;
+        return true;
       case DictTabInfo::ExtChar:
       case DictTabInfo::ExtBinary:
         AttributeType = DictTabInfo::StringType;
diff --git a/ndb/include/ndbapi/NdbDictionary.hpp b/ndb/include/ndbapi/NdbDictionary.hpp
index 49afbd695c9..6aa675a2319 100644
--- a/ndb/include/ndbapi/NdbDictionary.hpp
+++ b/ndb/include/ndbapi/NdbDictionary.hpp
@@ -179,7 +179,7 @@ public:
       Bigunsigned,   ///< 64 Bit. 8 byte signed integer, can be used in array
       Float,         ///< 32-bit float. 4 bytes float, can be used in array
       Double,        ///< 64-bit float. 8 byte float, can be used in array
-      Decimal,       ///< Precision, Scale are applicable
+      Olddecimal,    ///< MySQL < 5.0 signed decimal,  Precision, Scale
       Char,          ///< Len. A fixed array of 1-byte chars
       Varchar,       ///< Max len
       Binary,        ///< Len
@@ -190,7 +190,8 @@ public:
       Text,        ///< Text blob
       Time = 25,   ///< Time without date
       Year = 26,   ///< Year 1901-2155 (1 byte)
-      Timestamp = 27  ///< Unix time
+      Timestamp = 27, ///< Unix time
+      Olddecimalunsigned = 28
     };
 
     /** 
@@ -276,25 +277,25 @@ public:
 
     /**
      * Set precision of column.
-     * @note Only applicable for builtin type Decimal
+     * @note Only applicable for decimal types
      */
     void setPrecision(int);
 
     /**
      * Get precision of column.
-     * @note Only applicable for builtin type Decimal
+     * @note Only applicable for decimal types
      */
     int getPrecision() const;
 
     /**
      * Set scale of column.
-     * @note Only applicable for builtin type Decimal
+     * @note Only applicable for decimal types
      */
     void setScale(int);
 
     /**
      * Get scale of column.
-     * @note Only applicable for builtin type Decimal
+     * @note Only applicable for decimal types
      */
     int getScale() const;
 
diff --git a/ndb/include/util/NdbSqlUtil.hpp b/ndb/include/util/NdbSqlUtil.hpp
index 3787814052a..5b27bd4e0c4 100644
--- a/ndb/include/util/NdbSqlUtil.hpp
+++ b/ndb/include/util/NdbSqlUtil.hpp
@@ -75,7 +75,7 @@ public:
       Bigunsigned,      // 64 Bit
       Float,            // 32-bit float
       Double,           // 64-bit float
-      Decimal,          // Precision, Scale
+      Olddecimal,       // Precision, Scale
       Char,             // Len
       Varchar,          // Max len
       Binary,           // Len
@@ -86,7 +86,8 @@ public:
       Text,             // Text blob
       Time = 25,        // Time without date
       Year = 26,        // Year (size 1 byte)
-      Timestamp = 27    // Unix seconds (uint32)
+      Timestamp = 27,   // Unix seconds (uint32)
+      Olddecimalunsigned = 28
     };
     Enum m_typeId;
     Cmp* m_cmp;         // comparison method
@@ -109,6 +110,11 @@ public:
   static bool usable_in_hash_index(Uint32 typeId, const void* cs);
   static bool usable_in_ordered_index(Uint32 typeId, const void* cs);
 
+  /**
+   * Compare decimal numbers.
+   */
+  static int cmp_olddecimal(const uchar* s1, const uchar* s2, unsigned n);
+
 private:
   /**
    * List of all types.  Must match Type::Enum.
@@ -129,7 +135,7 @@ private:
   static Cmp cmpBigunsigned;
   static Cmp cmpFloat;
   static Cmp cmpDouble;
-  static Cmp cmpDecimal;
+  static Cmp cmpOlddecimal;
   static Cmp cmpChar;
   static Cmp cmpVarchar;
   static Cmp cmpBinary;
@@ -141,6 +147,7 @@ private:
   static Cmp cmpTime;
   static Cmp cmpYear;
   static Cmp cmpTimestamp;
+  static Cmp cmpOlddecimalunsigned;
 };
 
 #endif
diff --git a/ndb/src/common/util/NdbSqlUtil.cpp b/ndb/src/common/util/NdbSqlUtil.cpp
index 53fa5d69215..c4114ad5ffa 100644
--- a/ndb/src/common/util/NdbSqlUtil.cpp
+++ b/ndb/src/common/util/NdbSqlUtil.cpp
@@ -76,117 +76,121 @@ NdbSqlUtil::char_like(const char* s1, unsigned n1,
 
 const NdbSqlUtil::Type
 NdbSqlUtil::m_typeList[] = {
-  {
+  { // 0
     Type::Undefined,
     NULL
   },
-  {
+  { // 1
     Type::Tinyint,
     cmpTinyint
   },
-  {
+  { // 2
     Type::Tinyunsigned,
     cmpTinyunsigned
   },
-  {
+  { // 3
     Type::Smallint,
     cmpSmallint
   },
-  {
+  { // 4
     Type::Smallunsigned,
     cmpSmallunsigned
   },
-  {
+  { // 5
     Type::Mediumint,
     cmpMediumint
   },
-  {
+  { // 6
     Type::Mediumunsigned,
     cmpMediumunsigned
   },
-  {
+  { // 7
     Type::Int,
     cmpInt
   },
-  {
+  { // 8
     Type::Unsigned,
     cmpUnsigned
   },
-  {
+  { // 9
     Type::Bigint,
     cmpBigint
   },
-  {
+  { // 10
     Type::Bigunsigned,
     cmpBigunsigned
   },
-  {
+  { // 11
     Type::Float,
     cmpFloat
   },
-  {
+  { // 12
     Type::Double,
     cmpDouble
   },
-  {
-    Type::Decimal,
-    NULL  // cmpDecimal
+  { // 13
+    Type::Olddecimal,
+    cmpOlddecimal
   },
-  {
+  { // 14
     Type::Char,
     cmpChar
   },
-  {
+  { // 15
     Type::Varchar,
     cmpVarchar
   },
-  {
+  { // 16
     Type::Binary,
     cmpBinary
   },
-  {
+  { // 17
     Type::Varbinary,
     cmpVarbinary
   },
-  {
+  { // 18
     Type::Datetime,
     cmpDatetime
   },
-  {
+  { // 19
     Type::Date,
     cmpDate
   },
-  {
+  { // 20
     Type::Blob,
     cmpBlob
   },
-  {
+  { // 21
     Type::Text,
     cmpText
   },
-  {
+  { // 22
     Type::Undefined,    // 5.0 Bit
     NULL
   },
-  {
+  { // 23
     Type::Undefined,    // 5.0 Longvarchar
     NULL
   },
-  {
+  { // 24
     Type::Undefined,    // 5.0 Longvarbinary
     NULL
   },
-  {
+  { // 25
     Type::Time,
     cmpTime
   },
-  {
+  { // 26
     Type::Year,
     cmpYear
   },
-  {
+  { // 27
     Type::Timestamp,
     cmpTimestamp
+  },
+  { // 28
+    Type::Olddecimalunsigned,
+    cmpOlddecimalunsigned
   }
 };
 
@@ -411,15 +415,57 @@ NdbSqlUtil::cmpDouble(const void* info, const Uint32* p1, const Uint32* p2, Uint
 }
 
 int
-NdbSqlUtil::cmpDecimal(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
+NdbSqlUtil::cmp_olddecimal(const uchar* s1, const uchar* s2, unsigned n)
 {
-  assert(full >= size && size > 0);
-  // not used by MySQL or NDB
-  assert(false);
+  int sgn = +1;
+  unsigned i = 0;
+  while (i < n) {
+    int c1 = s1[i];
+    int c2 = s2[i];
+    if (c1 == c2) {
+      if (c1 == '-')
+        sgn = -1;
+    } else if (c1 == '-') {
+      return -1;
+    } else if (c2 == '-') {
+      return +1;
+    } else if (c1 < c2) {
+      return -1 * sgn;
+    } else {
+      return +1 * sgn;
+    }
+    i++;
+  }
   return 0;
 }
 
 int
+NdbSqlUtil::cmpOlddecimal(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
+{
+  assert(full >= size && size > 0);
+  if (full == size) {
+    union { const Uint32* p; const uchar* v; } u1, u2;
+    u1.p = p1;
+    u2.p = p2;
+    return cmp_olddecimal(u1.v, u2.v, full << 2);
+  }
+  return CmpUnknown;
+}
+
+int
+NdbSqlUtil::cmpOlddecimalunsigned(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
+{
+  assert(full >= size && size > 0);
+  if (full == size) {
+    union { const Uint32* p; const uchar* v; } u1, u2;
+    u1.p = p1;
+    u2.p = p2;
+    return cmp_olddecimal(u1.v, u2.v, full << 2);
+  }
+  return CmpUnknown;
+}
+
+int
 NdbSqlUtil::cmpChar(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
 {
   // collation does not work on prefix for some charsets
diff --git a/ndb/src/ndbapi/NdbDictionary.cpp b/ndb/src/ndbapi/NdbDictionary.cpp
index 58b35c6c306..00db5704949 100644
--- a/ndb/src/ndbapi/NdbDictionary.cpp
+++ b/ndb/src/ndbapi/NdbDictionary.cpp
@@ -918,8 +918,11 @@ operator<<(NdbOut& out, const NdbDictionary::Column& col)
   case NdbDictionary::Column::Double:
     out << "Double";
     break;
-  case NdbDictionary::Column::Decimal:
-    out << "Decimal(" << col.getScale() << "," << col.getPrecision() << ")";
+  case NdbDictionary::Column::Olddecimal:
+    out << "Olddecimal(" << col.getPrecision() << "," << col.getScale() << ")";
+    break;
+  case NdbDictionary::Column::Olddecimalunsigned:
+    out << "Olddecimalunsigned(" << col.getPrecision() << "," << col.getScale() << ")";
     break;
   case NdbDictionary::Column::Char:
     out << "Char(" << col.getLength() << ";" << csname << ")";
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index 530f15d3a2e..b9ae13a93ec 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -114,7 +114,8 @@ NdbColumnImpl::init(Type t)
     m_length = 1;
     m_cs = NULL;
     break;
-  case Decimal:
+  case Olddecimal:
+  case Olddecimalunsigned:
     m_precision = 10;
     m_scale = 0;
     m_length = 1;
@@ -1176,7 +1177,8 @@ columnTypeMapping[] = {
   { DictTabInfo::ExtBigunsigned,     NdbDictionary::Column::Bigunsigned },
   { DictTabInfo::ExtFloat,           NdbDictionary::Column::Float },
   { DictTabInfo::ExtDouble,          NdbDictionary::Column::Double },
-  { DictTabInfo::ExtDecimal,         NdbDictionary::Column::Decimal },
+  { DictTabInfo::ExtOlddecimal,      NdbDictionary::Column::Olddecimal },
+  { DictTabInfo::ExtOlddecimalunsigned, NdbDictionary::Column::Olddecimalunsigned },
   { DictTabInfo::ExtChar,            NdbDictionary::Column::Char },
   { DictTabInfo::ExtVarchar,         NdbDictionary::Column::Varchar },
   { DictTabInfo::ExtBinary,          NdbDictionary::Column::Binary },
diff --git a/ndb/src/ndbapi/NdbRecAttr.cpp b/ndb/src/ndbapi/NdbRecAttr.cpp
index 9c9a9cea8da..db83e9c5fcf 100644
--- a/ndb/src/ndbapi/NdbRecAttr.cpp
+++ b/ndb/src/ndbapi/NdbRecAttr.cpp
@@ -156,7 +156,8 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
     return out;
   }
 
-  uint length = r.getColumn()->getLength();
+  const NdbDictionary::Column* c = r.getColumn();
+  uint length = c->getLength();
   if (length > 1)
     out << "[";
 
@@ -208,6 +209,18 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
       case NdbDictionary::Column::Double:
 	out << r.double_value();
 	break;
+      case NdbDictionary::Column::Olddecimal:
+        {
+          short len = 1 + c->getPrecision() + (c->getScale() > 0);
+          out.print("%.*s", len, r.aRef());
+        }
+        break;
+      case NdbDictionary::Column::Olddecimalunsigned:
+        {
+          short len = 0 + c->getPrecision() + (c->getScale() > 0);
+          out.print("%.*s", len, r.aRef());
+        }
+	break;
       // for dates cut-and-paste from field.cc
       case NdbDictionary::Column::Datetime:
         {
diff --git a/ndb/test/include/NdbSchemaOp.hpp b/ndb/test/include/NdbSchemaOp.hpp
index 77e704c0e5c..e8ab542b00a 100644
--- a/ndb/test/include/NdbSchemaOp.hpp
+++ b/ndb/test/include/NdbSchemaOp.hpp
@@ -567,7 +567,8 @@ convertColumnTypeToAttrType(NdbDictionary::Column::Type _type)
   case NdbDictionary::Column::Unsigned:
     return UnSigned;
   case NdbDictionary::Column::Float:
-  case NdbDictionary::Column::Decimal:
+  case NdbDictionary::Column::Olddecimal:
+  case NdbDictionary::Column::Olddecimalunsigned:
   case NdbDictionary::Column::Double:
     return Float;
   case NdbDictionary::Column::Char:
diff --git a/ndb/tools/restore/consumer.cpp b/ndb/tools/restore/consumer.cpp
index 4d228230423..ecbdbbf8f4e 100644
--- a/ndb/tools/restore/consumer.cpp
+++ b/ndb/tools/restore/consumer.cpp
@@ -44,9 +44,12 @@ BackupConsumer::create_table_string(const TableS & table,
     case NdbDictionary::Column::Float:
       pos += sprintf(buf+pos, "%s", "float");
       break;
-    case NdbDictionary::Column::Decimal:
+    case NdbDictionary::Column::Olddecimal:
       pos += sprintf(buf+pos, "%s", "decimal");
       break;
+    case NdbDictionary::Column::Olddecimalunsigned:
+      pos += sprintf(buf+pos, "%s", "decimal unsigned");
+      break;
     case NdbDictionary::Column::Char:
       pos += sprintf(buf+pos, "%s", "char");
       break;