MCOL-4633 Remove duplicate code for DECIMAL to int64_t rounding conversion

Detailed change list:

- Splitting out the narrow part of "class Decimal" into a separate class TDecimal64

- Adding a method TDecimal64::toSInt64Round()

- Reusing the method TDecimal64::toSInt64Round() in:
   * Func_cast_signed::getIntVal()
   * Func_char::getStrVal()
   * Func_elt::getStrVal()
   * makedate()
   * Func_maketime::getStrVal()

   Note, reusing this method in Func_char::getStrVal() also fixed this bug:
     MCOL-4634 CHAR(negativeWideDecimal) is not like InnoDB
   because the old code handled negative wide decimal values
   in a wrong way.

- Adding a new class TDecimal128 for symmetry.
  Moving a few wide decimal methods and constexpr's from Decimal to TDecimal128.
  The new class TDecimal128 does not do much at this point yet.
  Later we should be able to use TDecimal128 vs TDecimal64 in templates.

datatypes/mcs_decimal.cpp

@@ -173,9 +173,7 @@ namespace datatypes
 
     Decimal::Decimal(const char *str, size_t length, DataCondition & convError,
                      int8_t s, uint8_t p)
-       :TSInt128(),
+       :scale(s),
-        value(0),
-        scale(s),
         precision(p)
     {
         literal::Converter<literal::SignedNumericLiteral> conv(str, length, convError);

datatypes/mcs_decimal.h

@@ -179,41 +179,87 @@ inline lldiv_t_128 lldiv128(const int128_t& dividend, const int128_t& divisor)
     return lldiv_t_128(dividend / divisor, dividend % divisor);
 }
 
+
+// TODO: derive it from TSInt64 eventually
+class TDecimal64
+{
+public:
+    int64_t value;
+public:
+    static constexpr uint8_t MAXLENGTH8BYTES = 23;
+
+public:
+    TDecimal64()
+     :value(0)
+    { }
+    explicit TDecimal64(int64_t val)
+     :value(val)
+    { }
+    // Divide to the scale divisor with rounding
+    int64_t toSInt64Round(uint32_t scale) const
+    {
+        double dscale = scale;
+        int64_t tmp = value / pow(10.0, dscale);
+        int lefto = (value - tmp * pow(10.0, dscale)) / pow(10.0, dscale - 1);
+        if (tmp >= 0 && lefto > 4)
+            return tmp + 1;
+        if (tmp < 0 && lefto < -4)
+            return tmp - 1;
+        return tmp;
+    }
+};
+
+
+class TDecimal128: public TSInt128
+{
+public:
+    static constexpr uint8_t MAXLENGTH16BYTES = TSInt128::maxLength();
+    static constexpr int128_t minInt128 = TFloat128::minInt128;
+    static constexpr int128_t maxInt128 = TFloat128::maxInt128;
+
+    static inline bool isWideDecimalNullValue(const int128_t& val)
+    {
+        return (val == TSInt128::NullValue);
+    }
+
+    static inline bool isWideDecimalEmptyValue(const int128_t& val)
+    {
+        return (val == TSInt128::EmptyValue);
+    }
+
+    static inline void setWideDecimalNullValue(int128_t& val)
+    {
+        val = TSInt128::NullValue;
+    }
+
+    static inline void setWideDecimalEmptyValue(int128_t& val)
+    {
+        val = TSInt128::EmptyValue;
+    }
+
+public:
+    TDecimal128()
+    { }
+    explicit TDecimal128(const int128_t &val)
+       :TSInt128(val)
+    { }
+    explicit TDecimal128(const TSInt128& val)
+       :TSInt128(val)
+    { }
+    explicit TDecimal128(const int128_t* valPtr)
+       :TSInt128(valPtr)
+    { }
+};
+
+
 // @brief The class for Decimal related operations
-// The class contains Decimal related operations are scale and
+// The methods and operators implemented in this class are
-// precision aware.
+// scale and precision aware.
-// This class will inherit from:
+// We should eventually move the members "scale" and "precision"
-//      DecimalMeta class that stores scale and precision
+// into a separate class DecimalMeta and derive Decimal from it.
-//      Storage classes TSInt128 and int64
+class Decimal: public TDecimal128, public TDecimal64
-// !!! There are some static classes that will exists during transition period. 
-class Decimal: public TSInt128
 {
     public:
-        static constexpr uint8_t MAXLENGTH16BYTES = TSInt128::maxLength();
-        static constexpr uint8_t MAXLENGTH8BYTES = 23;
-        static constexpr int128_t minInt128 = TFloat128::minInt128;
-        static constexpr int128_t maxInt128 = TFloat128::maxInt128;
-
-        static inline bool isWideDecimalNullValue(const int128_t& val)
-        {
-            return (val == TSInt128::NullValue);
-        }
-
-        static inline bool isWideDecimalEmptyValue(const int128_t& val)
-        {
-            return (val == TSInt128::EmptyValue);
-        }
-
-        static inline void setWideDecimalNullValue(int128_t& val)
-        {
-            val = TSInt128::NullValue;
-        }
-
-        static inline void setWideDecimalEmptyValue(int128_t& val)
-        {
-            val = TSInt128::EmptyValue;
-        }
-
         /**
             @brief Compares two Decimal taking scale into account. 
         */
@@ -277,27 +323,26 @@ class Decimal: public TSInt128
             precision += (precisionAvailable >= MAXSCALEINC4AVG) ? MAXSCALEINC4AVG : precisionAvailable;
         }
 
-        Decimal(): value(0), scale(0), precision(0)
+        Decimal(): scale(0), precision(0)
         {
         }
 
         Decimal(int64_t val, int8_t s, uint8_t p, const int128_t &val128 = 0) :
-            TSInt128(val128),
+            TDecimal128(val128),
-            value(val),
+            TDecimal64(val),
             scale(s),
             precision(p)
         { }
 
         Decimal(int64_t unused, int8_t s, uint8_t p, const int128_t* val128Ptr) :
-            TSInt128(val128Ptr),
+            TDecimal128(val128Ptr),
-            value(unused),
+            TDecimal64(unused),
             scale(s),
             precision(p)
         { }
 
         Decimal(const TSInt128& val128, int8_t s, uint8_t p) :
-            TSInt128(val128),
+            TDecimal128(val128),
-            value(0),
             scale(s),
             precision(p)
         { }
@@ -481,6 +526,13 @@ class Decimal: public TSInt128
             return intg;
         }
 
+        int64_t toSInt64Round() const
+        {
+            return isWideDecimalTypeByPrecision(precision) ?
+                static_cast<int64_t>(getPosNegRoundedIntegralPart(4)) :
+                TDecimal64::toSInt64Round((uint32_t) scale);
+        }
+
         // MOD operator for an integer divisor to be used
         // for integer rhs
         inline TSInt128 operator%(const TSInt128& div) const
@@ -627,7 +679,6 @@ class Decimal: public TSInt128
         std::string toString(bool hasTSInt128 = false) const;
         friend std::ostream& operator<<(std::ostream& os, const Decimal& dec);
 
-        int64_t value;
         int8_t  scale;	  // 0~38
         uint8_t precision;  // 1~38
 

mtr/basic/r/type_decimal.result

@@ -3547,3 +3547,18 @@ a	CAST(a AS DECIMAL(38,0))
 .00000000000000000000000000000000000000e-111111111111111111111	0
 .00000000000000000000000000000000000001e-111111111111111111111	0
 DROP TABLE t1;
+#
+# MCOL-4604 CHAR(negativeWideDecimal) not behaving as InnoDB
+#
+CREATE TABLE t1 (a DECIMAL(10,0));
+INSERT INTO t1 VALUES (-1);
+SELECT HEX(CHAR(a USING latin1)) FROM t1;
+HEX(CHAR(a USING latin1))
+FFFFFFFF
+DROP TABLE t1;
+CREATE TABLE t1 (a DECIMAL(30,0));
+INSERT INTO t1 VALUES (-1);
+SELECT HEX(CHAR(a USING latin1)) FROM t1;
+HEX(CHAR(a USING latin1))
+FFFFFFFF
+DROP TABLE t1;

mtr/basic/t/type_decimal.test

@@ -192,3 +192,18 @@ INSERT INTO t1 VALUES
 SELECT a, CAST(a AS DECIMAL(38,0)) FROM t1 ORDER BY LENGTH(a), a;
 --enable_warnings
 DROP TABLE t1;
+
+
+--echo #
+--echo # MCOL-4604 CHAR(negativeWideDecimal) not behaving as InnoDB
+--echo #
+
+CREATE TABLE t1 (a DECIMAL(10,0));
+INSERT INTO t1 VALUES (-1);
+SELECT HEX(CHAR(a USING latin1)) FROM t1;
+DROP TABLE t1;
+
+CREATE TABLE t1 (a DECIMAL(30,0));
+INSERT INTO t1 VALUES (-1);
+SELECT HEX(CHAR(a USING latin1)) FROM t1;
+DROP TABLE t1;

utils/funcexp/func_cast.cpp

@@ -183,26 +183,7 @@ int64_t Func_cast_signed::getIntVal(Row& row,
         case execplan::CalpontSystemCatalog::DECIMAL:
         case execplan::CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal d = parm[0]->data()->getDecimalVal(row, isNull);
+            return parm[0]->data()->getDecimalVal(row, isNull).toSInt64Round();
-
-            if (parm[0]->data()->resultType().colWidth == datatypes::MAXDECIMALWIDTH)
-            {
-                return static_cast<int64_t>(d.getPosNegRoundedIntegralPart(4));
-            }
-            else
-            {
-                double dscale = d.scale;
-                int64_t value = d.value / pow(10.0, dscale);
-                int lefto = (d.value - value * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                if ( value >= 0 && lefto > 4 )
-                    value++;
-
-                if ( value < 0 && lefto < -4 )
-                    value--;
-
-                return value;
-            }
         }
         break;
 

utils/funcexp/func_char.cpp

@@ -136,26 +136,7 @@ string Func_char::getStrVal(Row& row,
             case execplan::CalpontSystemCatalog::DECIMAL:
             case execplan::CalpontSystemCatalog::UDECIMAL:
             {
-                IDB_Decimal d = rc->getDecimalVal(row, isNull);
+                value = static_cast<int32_t>(rc->getDecimalVal(row, isNull).toSInt64Round());
-
-                uint8_t roundingFactor = 4;
-                if (ct.colWidth == datatypes::MAXDECIMALWIDTH)
-                {
-                    if (d.s128Value < 0)
-                        return "";
-                    // rounding by the left over
-                    value = static_cast<int32_t>(d.getRoundedIntegralPart(roundingFactor));
-                }
-                else
-                {
-                    double dscale = d.scale;
-                    // get decimal and round up
-                    value = d.value / pow(10.0, dscale);
-                    uint8_t lefto = (d.value - value * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                    if ( lefto > roundingFactor )
-                        value++;
-                }
             }
             break;
 

utils/funcexp/func_elt.cpp

@@ -73,25 +73,7 @@ string Func_elt::getStrVal(rowgroup::Row& row,
         case CalpontSystemCatalog::DECIMAL:
         case CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal d = parm[0]->data()->getDecimalVal(row, isNull);
+            number = static_cast<uint64_t>(parm[0]->data()->getDecimalVal(row, isNull).toSInt64Round());
-
-            if (parm[0]->data()->resultType().colWidth == datatypes::MAXDECIMALWIDTH)
-            {
-                number = static_cast<int64_t>(d.getPosNegRoundedIntegralPart(4));
-            }
-            else
-            {
-                double dscale = d.scale;
-                number = d.value / pow(10.0, dscale);
-                int lefto = (d.value - number * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                if ( utils::is_nonnegative(number) && lefto > 4 )
-                    number++;
-
-                if ( utils::is_negative(number) && lefto < -4 )
-                    number--;
-            }
-
             break;
         }
 

utils/funcexp/func_makedate.cpp

@@ -68,25 +68,7 @@ uint64_t makedate(rowgroup::Row& row,
         case CalpontSystemCatalog::DECIMAL:
         case CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal d = parm[0]->data()->getDecimalVal(row, isNull);
+            year = parm[0]->data()->getDecimalVal(row, isNull).toSInt64Round();
-
-            if (parm[0]->data()->resultType().colWidth == datatypes::MAXDECIMALWIDTH)
-            {
-                year = static_cast<int64_t>(d.getPosNegRoundedIntegralPart(4));
-            }
-            else
-            {
-                double dscale = d.scale;
-                year = d.value / pow(10.0, dscale);
-                int lefto = (d.value - year * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                if ( year >= 0 && lefto > 4 )
-                    year++;
-
-                if ( year < 0 && lefto < -4 )
-                    year--;
-            }
-
             break;
         }
 
@@ -137,39 +119,13 @@ uint64_t makedate(rowgroup::Row& row,
         case CalpontSystemCatalog::DECIMAL:
         case CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal d = parm[1]->data()->getDecimalVal(row, isNull);
+            int64_t tmp = parm[1]->data()->getDecimalVal(row, isNull).toSInt64Round();
-
+            if (tmp < 1)
-            if (parm[1]->data()->resultType().colWidth == datatypes::MAXDECIMALWIDTH)
             {
-                int64_t tmpval = static_cast<int64_t>(d.getPosNegRoundedIntegralPart(4));
+                isNull = true;
-                if (tmpval < 1)
+                return 0;
-                {
-                    isNull = true;
-                    return 0;
-                }
-
-                dayofyear = helpers::intToString(tmpval);
-            }
-            else
-            {
-                double dscale = d.scale;
-                int64_t tmp = d.value / pow(10.0, dscale);
-                int lefto = (d.value - tmp * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                if (tmp >= 0 && lefto > 4)
-                    tmp++;
-
-                if (tmp < 0 && lefto < -4)
-                    tmp--;
-
-                if (tmp < 1)
-                {
-                    isNull = true;
-                    return 0;
-                }
-
-                dayofyear = helpers::intToString(tmp);
             }
+            dayofyear = helpers::intToString(tmp);
             break;
         }
 

utils/funcexp/func_maketime.cpp

@@ -74,25 +74,7 @@ string Func_maketime::getStrVal(rowgroup::Row& row,
         case CalpontSystemCatalog::DECIMAL:
         case CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal d = parm[0]->data()->getDecimalVal(row, isNull);
+            hour = parm[0]->data()->getDecimalVal(row, isNull).toSInt64Round();
-
-            if (parm[0]->data()->resultType().colWidth == datatypes::MAXDECIMALWIDTH)
-            {
-                hour = static_cast<int64_t>(d.getPosNegRoundedIntegralPart(4));
-            }
-            else
-            {
-                double dscale = d.scale;
-                hour = d.value / pow(10.0, dscale);
-                int lefto = (d.value - hour * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                if ( hour >= 0 && lefto > 4 )
-                    hour++;
-
-                if ( hour < 0 && lefto < -4 )
-                    hour--;
-            }
-
             break;
         }
 
@@ -123,25 +105,7 @@ string Func_maketime::getStrVal(rowgroup::Row& row,
         case CalpontSystemCatalog::DECIMAL:
         case CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal d = parm[1]->data()->getDecimalVal(row, isNull);
+            min = parm[1]->data()->getDecimalVal(row, isNull).toSInt64Round();
-
-            if (parm[1]->data()->resultType().colWidth == datatypes::MAXDECIMALWIDTH)
-            {
-                min = static_cast<int64_t>(d.getPosNegRoundedIntegralPart(4));
-            }
-            else
-            {
-                double dscale = d.scale;
-                min = d.value / pow(10.0, dscale);
-                int lefto = (d.value - min * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                if ( min >= 0 && lefto > 4 )
-                    min++;
-
-                if ( min < 0 && lefto < -4 )
-                    min--;
-            }
-
             break;
         }
 
@@ -178,25 +142,7 @@ string Func_maketime::getStrVal(rowgroup::Row& row,
         case CalpontSystemCatalog::DECIMAL:
         case CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal d = parm[2]->data()->getDecimalVal(row, isNull);
+            sec = parm[2]->data()->getDecimalVal(row, isNull).toSInt64Round();
-
-            if (parm[2]->data()->resultType().colWidth == datatypes::MAXDECIMALWIDTH)
-            {
-                sec = static_cast<int64_t>(d.getPosNegRoundedIntegralPart(4));
-            }
-            else
-            {
-                double dscale = d.scale;
-                sec = d.value / pow(10.0, dscale);
-                int lefto = (d.value - sec * pow(10.0, dscale)) / pow(10.0, dscale - 1);
-
-                if ( sec >= 0 && lefto > 4 )
-                    sec++;
-
-                if ( sec < 0 && lefto < -4 )
-                    sec--;
-            }
-
             break;
         }