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

- MCOL-4618 FLOOR(-9999.0) returns a bad result
- MCOL-4653 CEIL(negativeNarrowDecimal) returns a wrong result

Main changes:
a. Moving ROUND, CEIL, FLOOR related code into a new simple
   class template DecomposedDecimal, which is reused for
   64 and 128 bit decimal.
b. Using DecomposedDecimal in TDecimal64 and TDecimal128
   to implement ROUND, CEIL, FLOOR related methods.
c. Adding corresponding wrapper methods to the class Decimal.
d. Using new Decimal methods in Func_ceil and Func_floor.

Additional minor changed:
- Adding "explicit" to TSInt128 constructors to avoid
  hidden data type conversion and erroneous choice between
  64 vs 128 bit APIs when using Decimal.
  Now one can call constructors in this self explanatory way:
  - Decimal(TSInt128(some_int_value), scale, precision)
    to create a wide decimal
  - Decimal(TSInt64(some_int_value, scale, precision)
    to create a narrow decimal

  TODO:
  Consider changing
    Decimal(int64_t val, int8_t s, uint8_t p, const int128_t &val128 = 0)
  to
    Decimal(int64_t val, int8_t s, uint8_t p, const int128_t &val128)
  (or even removing this constructor) to disallow compilation of:
    Decimal(some_trivial_type_value, scale, precision)

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;