Merge pull request #1836 from mariadb-corporation/bar-develop-MCOL-4618

A joint patch fixing MCOL-4618 and MCOL-4653:

datatypes/mcs_decimal.h

@@ -26,6 +26,7 @@
 #include "exceptclasses.h"
 #include "widedecimalutils.h"
 #include "mcs_int128.h"
+#include "mcs_int64.h"
 #include "mcs_float128.h"
 #include "checks.h"
 #include "branchpred.h"
@@ -153,6 +154,47 @@ T scaleDivisor(const uint32_t scale)
     return (T) mcs_pow_10_128[scale - 19];
 }
 
+
+// Decomposed Decimal representation
+// T - storage data type (int64_t, int128_t)
+template<typename T>class DecomposedDecimal
+{
+    T mDivisor;
+    T mIntegral;
+    T mFractional;
+public:
+    DecomposedDecimal(T value, uint32_t scale)
+        :mDivisor(scaleDivisor<T>(scale)),
+         mIntegral(value / mDivisor),
+         mFractional(value % mDivisor)
+    { }
+    T toSIntRound() const
+    {
+        T frac2 = 2 * mFractional;
+        if (frac2 >= mDivisor)
+            return mIntegral + 1;
+        if (frac2 <= -mDivisor)
+            return mIntegral - 1;
+        return mIntegral;
+    }
+    T toSIntRoundPositive() const
+    {
+        T frac2 = 2 * mFractional;
+        if (frac2 >= mDivisor)
+            return mIntegral + 1;
+        return mIntegral;
+    }
+    T toSIntFloor() const
+    {
+      return mFractional < 0 ? mIntegral - 1 : mIntegral;
+    }
+    T toSIntCeil() const
+    {
+      return mFractional > 0 ? mIntegral + 1 : mIntegral;
+    }
+};
+
+
 /**
     @brief The function to produce scale multiplier/divisor for
     wide decimals.
@@ -201,17 +243,13 @@ public:
     explicit TDecimal64(int64_t val)
      :value(val)
     { }
+    explicit TDecimal64(const TSInt64 &val)
+     :value(static_cast<int64_t>(val))
+    { }
     // Divide to the scale divisor with rounding
     int64_t toSInt64Round(uint32_t scale) const
     {
-        auto divisor = scaleDivisor<int64_t>(scale);
+        return DecomposedDecimal<int64_t>(value, scale).toSIntRound();
-        int64_t intg = value / divisor;
-        int64_t frac2 = 2 * (value % divisor);
-        if (frac2 >= divisor)
-          return intg + 1;
-        if (frac2 <= -divisor)
-          return intg - 1;
-        return intg;
     }
     uint64_t toUInt64Round(uint32_t scale) const
     {
@@ -219,6 +257,16 @@ public:
                0 :
                static_cast<uint64_t>(toSInt64Round(scale));
     }
+
+    int64_t toSInt64Floor(uint32_t scale) const
+    {
+        return DecomposedDecimal<int64_t>(value, scale).toSIntFloor();
+    }
+
+    int64_t toSInt64Ceil(uint32_t scale) const
+    {
+        return DecomposedDecimal<int64_t>(value, scale).toSIntCeil();
+    }
 };
 
 
@@ -265,14 +313,21 @@ public:
     {
         if (s128Value <= 0)
             return 0;
-        auto divisor = scaleDivisor<uint128_t>(scale);
+        int128_t intg = DecomposedDecimal<int128_t>(s128Value, scale).
-        uint128_t intg = s128Value / divisor;
+                          toSIntRoundPositive();
-        uint128_t frac2 = 2 * (s128Value % divisor);
-        if (frac2 >= divisor)
-            intg++;
         return intg > numeric_limits<uint64_t>::max() ? numeric_limits<uint64_t>::max() :
                                                         static_cast<uint64_t>(intg);
     }
+
+    int128_t toSInt128Floor(uint32_t scale) const
+    {
+        return DecomposedDecimal<int128_t>(s128Value, scale).toSIntFloor();
+    }
+
+    int128_t toSInt128Ceil(uint32_t scale) const
+    {
+        return DecomposedDecimal<int128_t>(s128Value, scale).toSIntCeil();
+    }
 };
 
 
@@ -358,6 +413,12 @@ class Decimal: public TDecimal128, public TDecimal64
             precision(p)
         { }
 
+        Decimal(const TSInt64 &val, int8_t s, uint8_t p) :
+            TDecimal64(val),
+            scale(s),
+            precision(p)
+        { }
+
         Decimal(int64_t unused, int8_t s, uint8_t p, const int128_t* val128Ptr) :
             TDecimal128(val128Ptr),
             TDecimal64(unused),
@@ -502,23 +563,6 @@ class Decimal: public TDecimal128, public TDecimal64
             return TSInt128(getIntegralPartNegativeScale(scaleDivisor));
         }
 
-        // !!! This is a very hevyweight rouding style
-        // Argument determines if it is a ceil
-        // rounding or not. 0 - ceil rounding
-        inline TSInt128 getRoundedIntegralPart(const uint8_t roundingFactor = 0) const
-        {
-            int128_t flooredScaleDivisor = 0;
-            int128_t roundedValue = getIntegralPartNonNegativeScale(flooredScaleDivisor);
-            int128_t ceiledScaleDivisor = (flooredScaleDivisor <= 10) ? 1 : (flooredScaleDivisor / 10);
-            int128_t leftO = (s128Value - roundedValue * flooredScaleDivisor) / ceiledScaleDivisor;
-            if (leftO > roundingFactor)
-            {
-                roundedValue++;
-            }
-
-            return TSInt128(roundedValue);
-        }
-
         inline TSInt128 getPosNegRoundedIntegralPart(const uint8_t roundingFactor = 0) const
         {
             int128_t flooredScaleDivisor = 0;
@@ -559,13 +603,57 @@ class Decimal: public TDecimal128, public TDecimal64
                    TDecimal64::toUInt64Round((uint32_t) scale);
         }
 
+        // FLOOR related routines
+        int64_t toSInt64Floor() const
+        {
+            return isWideDecimalTypeByPrecision(precision) ?
+              static_cast<int64_t>(TSInt128(TDecimal128::toSInt128Floor((uint32_t) scale))) :
+              TDecimal64::toSInt64Floor((uint32_t) scale);
+        }
+
+        uint64_t toUInt64Floor() const
+        {
+            return isWideDecimalTypeByPrecision(precision) ?
+              static_cast<uint64_t>(TSInt128(TDecimal128::toSInt128Floor((uint32_t) scale))) :
+              static_cast<uint64_t>(TSInt64(TDecimal64::toSInt64Floor((uint32_t) scale)));
+        }
+
+        Decimal floor() const
+        {
+            return isWideDecimalTypeByPrecision(precision)?
+              Decimal(TSInt128(TDecimal128::toSInt128Floor((uint32_t) scale)), 0, precision) :
+              Decimal(TSInt64(TDecimal64::toSInt64Floor((uint32_t) scale)), 0, precision);
+        }
+
+        // CEIL related routines
+        int64_t toSInt64Ceil() const
+        {
+            return isWideDecimalTypeByPrecision(precision) ?
+              static_cast<int64_t>(TSInt128(TDecimal128::toSInt128Ceil((uint32_t) scale))) :
+              static_cast<int64_t>(TSInt64(TDecimal64::toSInt64Ceil((uint32_t) scale)));
+        }
+
+        uint64_t toUInt64Ceil() const
+        {
+            return isWideDecimalTypeByPrecision(precision) ?
+              static_cast<uint64_t>(TSInt128(TDecimal128::toSInt128Ceil((uint32_t) scale))) :
+              static_cast<uint64_t>(TSInt64(TDecimal64::toSInt64Ceil((uint32_t) scale)));
+        }
+
+        Decimal ceil() const
+        {
+            return isWideDecimalTypeByPrecision(precision) ?
+              Decimal(TSInt128(TDecimal128::toSInt128Ceil((uint32_t) scale)), 0, precision) :
+              Decimal(TSInt64(TDecimal64::toSInt64Ceil((uint32_t) scale)), 0, precision);
+        }
+
         // MOD operator for an integer divisor to be used
         // for integer rhs
         inline TSInt128 operator%(const TSInt128& div) const
         {
             if (!isScaled())
             {
-                return s128Value % div.getValue();
+                return TSInt128(s128Value % div.getValue());
             }
             // Scale the value and calculate
             // (LHS.value % RHS.value) * LHS.scaleMultiplier + LHS.scale_div_remainder

datatypes/mcs_int128.h

@@ -144,13 +144,13 @@ class TSInt128
     TSInt128(const TSInt128& other): s128Value(other.s128Value) { }
 
     //    aligned argument
-    TSInt128(const int128_t& x) { s128Value = x; }
+    explicit TSInt128(const int128_t& x) { s128Value = x; }
 
     //    unaligned argument
-    TSInt128(const int128_t* x) { assignPtrPtr(&s128Value, x); }
+    explicit TSInt128(const int128_t* x) { assignPtrPtr(&s128Value, x); }
 
     //    unaligned argument
-    TSInt128(const unsigned char* x) { assignPtrPtr(&s128Value, x); }
+    explicit TSInt128(const unsigned char* x) { assignPtrPtr(&s128Value, x); }
 
     //    Method returns max length of a string representation
     static constexpr uint8_t maxLength()

datatypes/mcs_int64.h

@@ -67,6 +67,10 @@ public:
   {
     return mValue;
   }
+  explicit operator uint64_t () const
+  {
+    return mValue < 0 ? 0 : static_cast<uint64_t>(mValue);
+  }
 };
 
 

mtr/basic/r/type_decimal.result

@@ -3705,3 +3705,64 @@ a	SEC_TO_TIME(a)
 18446744073709551615.0	838:59:59.9
 18446744073709551615.5	838:59:59.9
 DROP TABLE t1;
+#
+# MCOL-4618 FLOOR(-9999.0) returns a bad result
+#
+CREATE TABLE t1 (d DECIMAL(18,1));
+INSERT INTO t1 VALUES (-9999.0);
+SELECT  d, FLOOR(d) FROM t1;
+d	FLOOR(d)
+-9999.0	-9999
+DROP TABLE t1;
+#
+# MCOL-4653 CEIL(negativeNarrowDecimal) returns a wrong result
+#
+CREATE TABLE t1 (a DECIMAL(10,1));
+INSERT INTO t1 VALUES (-1.6);
+SELECT * FROM t1 WHERE LEFT('abc', CEIL(a))='';
+a
+-1.6
+DROP TABLE t1;
+#
+# General FLOOR, CEIL, ROUND tests
+#
+CREATE TABLE t1 (d DECIMAL(18,1));
+INSERT INTO t1 VALUES (-9999.0),(-9999.4),(-9999.5);
+INSERT INTO t1 VALUES (9999.0),(9999.4),(9999.5);
+SELECT  d, FLOOR(d), ROUND(d), CEIL(d) FROM t1 ORDER BY d;
+d	FLOOR(d)	ROUND(d)	CEIL(d)
+-9999.5	-10000	-10000	-9999
+-9999.4	-10000	-9999	-9999
+-9999.0	-9999	-9999	-9999
+9999.0	9999	9999	9999
+9999.4	9999	9999	10000
+9999.5	9999	10000	10000
+SELECT d, FLOOR(d)+0.0, ROUND(d)+0.0, CEIL(d)+0.0 FROM t1 ORDER BY d;
+d	FLOOR(d)+0.0	ROUND(d)+0.0	CEIL(d)+0.0
+-9999.5	-10000.0	-10000.0	-9999.0
+-9999.4	-10000.0	-9999.0	-9999.0
+-9999.0	-9999.0	-9999.0	-9999.0
+9999.0	9999.0	9999.0	9999.0
+9999.4	9999.0	9999.0	10000.0
+9999.5	9999.0	10000.0	10000.0
+DROP TABLE t1;
+CREATE TABLE t1 (d DECIMAL(30,1));
+INSERT INTO t1 VALUES (-9999.0),(-9999.4),(-9999.5);
+INSERT INTO t1 VALUES (9999.0),(9999.4),(9999.5);
+SELECT  d, FLOOR(d), ROUND(d), CEIL(d) FROM t1 ORDER BY d;
+d	FLOOR(d)	ROUND(d)	CEIL(d)
+-9999.5	-10000	-10000	-9999
+-9999.4	-10000	-9999	-9999
+-9999.0	-9999	-9999	-9999
+9999.0	9999	9999	9999
+9999.4	9999	9999	10000
+9999.5	9999	10000	10000
+SELECT d, FLOOR(d)+0.0, ROUND(d)+0.0, CEIL(d)+0.0 FROM t1 ORDER BY d;
+d	FLOOR(d)+0.0	ROUND(d)+0.0	CEIL(d)+0.0
+-9999.5	-10000.0	-10000.0	-9999.0
+-9999.4	-10000.0	-9999.0	-9999.0
+-9999.0	-9999.0	-9999.0	-9999.0
+9999.0	9999.0	9999.0	9999.0
+9999.4	9999.0	9999.0	10000.0
+9999.5	9999.0	10000.0	10000.0
+DROP TABLE t1;

mtr/basic/t/type_decimal.test

@@ -334,3 +334,42 @@ INSERT INTO t1 VALUES (18446744073709551615.5);
 SELECT a, SEC_TO_TIME(a) FROM t1 ORDER BY a;
 --enable_warnings
 DROP TABLE t1;
+
+
+--echo #
+--echo # MCOL-4618 FLOOR(-9999.0) returns a bad result
+--echo #
+
+CREATE TABLE t1 (d DECIMAL(18,1));
+INSERT INTO t1 VALUES (-9999.0);
+SELECT  d, FLOOR(d) FROM t1;
+DROP TABLE t1;
+
+--echo #
+--echo # MCOL-4653 CEIL(negativeNarrowDecimal) returns a wrong result
+--echo #
+
+CREATE TABLE t1 (a DECIMAL(10,1));
+INSERT INTO t1 VALUES (-1.6);
+# ColumnStore returns NULL. Other engines return empty strings.
+# This condition works for ColumnStore and for all other engines:
+SELECT * FROM t1 WHERE LEFT('abc', CEIL(a))='';
+DROP TABLE t1;
+
+--echo #
+--echo # General FLOOR, CEIL, ROUND tests
+--echo #
+
+CREATE TABLE t1 (d DECIMAL(18,1));
+INSERT INTO t1 VALUES (-9999.0),(-9999.4),(-9999.5);
+INSERT INTO t1 VALUES (9999.0),(9999.4),(9999.5);
+SELECT  d, FLOOR(d), ROUND(d), CEIL(d) FROM t1 ORDER BY d;
+SELECT d, FLOOR(d)+0.0, ROUND(d)+0.0, CEIL(d)+0.0 FROM t1 ORDER BY d;
+DROP TABLE t1;
+
+CREATE TABLE t1 (d DECIMAL(30,1));
+INSERT INTO t1 VALUES (-9999.0),(-9999.4),(-9999.5);
+INSERT INTO t1 VALUES (9999.0),(9999.4),(9999.5);
+SELECT  d, FLOOR(d), ROUND(d), CEIL(d) FROM t1 ORDER BY d;
+SELECT d, FLOOR(d)+0.0, ROUND(d)+0.0, CEIL(d)+0.0 FROM t1 ORDER BY d;
+DROP TABLE t1;

utils/funcexp/func_ceil.cpp

@@ -97,22 +97,7 @@ int64_t Func_ceil::getIntVal(Row& row,
                     throw logging::IDBExcept(oss.str(), ERR_DATATYPE_NOT_SUPPORT);
                 }
 
-                if (op_ct.colWidth == datatypes::MAXDECIMALWIDTH)
+                ret = d.toSInt64Ceil();
-                {
-                    ret = static_cast<int64_t>(d.getRoundedIntegralPart());
-                }
-                else
-                {
-                    int64_t tmp = d.value;
-                    d.value /= helpers::powerOf10_c[d.scale];
-
-                    // Add 1 if this is a positive number and there were values to the right of the
-                    // decimal point so that we return the largest integer value not less than X.
-                    if ((tmp - (d.value * helpers::powerOf10_c[d.scale])) > 0)
-                        d.value += 1;
-
-                    ret = d.value;
-                }
             }
         }
         break;
@@ -240,22 +225,7 @@ uint64_t Func_ceil::getUintVal(Row& row,
                     throw logging::IDBExcept(oss.str(), ERR_DATATYPE_NOT_SUPPORT);
                 }
 
-                if (op_ct.colWidth == datatypes::MAXDECIMALWIDTH)
+                ret = d.toUInt64Ceil();
-                {
-                    ret = static_cast<uint64_t>(d.getRoundedIntegralPart());
-                }
-                else
-                {
-                    int64_t tmp = d.value;
-                    d.value /= helpers::powerOf10_c[d.scale];
-
-                    // Add 1 if this is a positive number and there were values to the right of the
-                    // decimal point so that we return the largest integer value not less than X.
-                    if ((tmp - (d.value * helpers::powerOf10_c[d.scale])) > 0)
-                        d.value += 1;
-
-                    ret = (uint64_t) d.value;
-                }
             }
         }
         break;
@@ -562,23 +532,7 @@ IDB_Decimal Func_ceil::getDecimalVal(Row& row,
                         << " with scale " << (int) ret.scale << " is beyond supported scale";
                     throw logging::IDBExcept(oss.str(), ERR_DATATYPE_NOT_SUPPORT);
                 }
-
+                return ret.ceil();
-                if (op_ct.colWidth == datatypes::MAXDECIMALWIDTH)
-                {
-                    ret = IDB_Decimal(ret.getRoundedIntegralPart(),
-                                      ret.scale,
-                                      ret.precision);
-                }
-                else
-                {
-                    int64_t tmp = ret.value;
-                    ret.value /= helpers::powerOf10_c[ret.scale];
-
-                    // Add 1 if this is a positive number and there were values to the right of the
-                    // decimal point so that we return the largest integer value not less than X.
-                    if ((tmp - (ret.value * helpers::powerOf10_c[ret.scale])) > 0)
-                        ret.value += 1;
-                }
             }
         }
         break;

utils/funcexp/func_floor.cpp

@@ -138,16 +138,7 @@ int64_t Func_floor::getIntVal(Row& row,
         case CalpontSystemCatalog::DECIMAL:
         case CalpontSystemCatalog::UDECIMAL:
         {
-            IDB_Decimal tmp = getDecimalVal(row, parm, isNull, op_ct);
+            ret = parm[0]->data()->getDecimalVal(row, isNull).toSInt64Floor();
-
-            if (op_ct.colWidth == datatypes::MAXDECIMALWIDTH)
-            {
-                ret = static_cast<int64_t>(tmp.toTSInt128());
-            }
-            else
-            {
-                ret = tmp.value;
-            }
             break;
         }
 
@@ -177,8 +168,6 @@ uint64_t Func_floor::getUintVal(Row& row,
         case execplan::CalpontSystemCatalog::MEDINT:
         case execplan::CalpontSystemCatalog::TINYINT:
         case execplan::CalpontSystemCatalog::SMALLINT:
-        case execplan::CalpontSystemCatalog::DECIMAL:
-        case execplan::CalpontSystemCatalog::UDECIMAL:
         {
             ret = parm[0]->data()->getIntVal(row, isNull);
         }
@@ -268,6 +257,13 @@ uint64_t Func_floor::getUintVal(Row& row,
         }
         break;
 
+        case CalpontSystemCatalog::DECIMAL:
+        case CalpontSystemCatalog::UDECIMAL:
+        {
+            ret = parm[0]->data()->getDecimalVal(row, isNull).toUInt64Floor();
+            break;
+        }
+
         default:
         {
             std::ostringstream oss;
@@ -483,26 +479,7 @@ IDB_Decimal Func_floor::getDecimalVal(Row& row,
                     throw logging::IDBExcept(oss.str(), ERR_DATATYPE_NOT_SUPPORT);
                 }
 
-                if (op_ct.colWidth == datatypes::MAXDECIMALWIDTH)
+                return ret.floor();
-                {
-                    int128_t tmp = ret.s128Value;
-                    int128_t scaleDivisor;
-                    datatypes::getScaleDivisor(scaleDivisor, ret.scale);
-                    ret.s128Value /= scaleDivisor;
-
-                    // Largest integer value not greater than X.
-                    if (tmp < 0 && tmp < ret.s128Value)
-                        ret.s128Value -= 1;
-                }
-                else
-                {
-                    int64_t tmp = ret.value;
-                    ret.value /= helpers::powerOf10_c[ret.scale];
-
-                    // Largest integer value not greater than X.
-                    if (tmp < 0 && tmp < ret.value)
-                        ret.value -= 1;
-                }
             }
         }
         break;