MCOL-641 Implement int128_t versions of arithmetic operations and add unit test cases.

datatypes/mcs_decimal.cpp

@@ -48,7 +48,7 @@ namespace datatypes
     template<typename BinaryOperation,
         typename OpOverflowCheck,
         typename MultiplicationOverflowCheck>
-    void execute(const execplan::IDB_Decimal& l,
+    void addSubtractExecute(const execplan::IDB_Decimal& l,
         const execplan::IDB_Decimal& r,
         execplan::IDB_Decimal& result,
         BinaryOperation op,
@@ -57,7 +57,7 @@ namespace datatypes
     {
         int128_t lValue = Decimal::isWideDecimalType(l.precision)
             ? l.s128Value : l.value;
-        int128_t rValue = Decimal::isWideDecimalType(l.precision)
+        int128_t rValue = Decimal::isWideDecimalType(r.precision)
             ? r.s128Value : r.value;
 
         if (result.scale == l.scale && result.scale == r.scale)
@@ -78,7 +78,9 @@ namespace datatypes
         {
             int128_t scaleMultiplier;
             getScaleDivisor(scaleMultiplier, l.scale - result.scale);
-            lValue /= scaleMultiplier;
+            lValue = (int128_t) (lValue > 0 ?
+                                 (__float128)lValue / scaleMultiplier + 0.5 :
+                                 (__float128)lValue / scaleMultiplier - 0.5);
         }
 
         if (result.scale > r.scale)
@@ -92,7 +94,9 @@ namespace datatypes
         {
             int128_t scaleMultiplier;
             getScaleDivisor(scaleMultiplier, r.scale - result.scale);
-            rValue /= scaleMultiplier;
+            rValue = (int128_t) (rValue > 0 ?
+                                 (__float128)rValue / scaleMultiplier + 0.5 :
+                                 (__float128)rValue / scaleMultiplier - 0.5);
         }
 
         // We assume there is no way that lValue or rValue calculations
@@ -102,6 +106,94 @@ namespace datatypes
         result.s128Value = op(lValue, rValue);
     }
 
+    template<typename OpOverflowCheck,
+        typename MultiplicationOverflowCheck>
+    void divisionExecute(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r,
+        execplan::IDB_Decimal& result,
+        OpOverflowCheck opOverflowCheck,
+        MultiplicationOverflowCheck mulOverflowCheck)
+    {
+        int128_t lValue = Decimal::isWideDecimalType(l.precision)
+            ? l.s128Value : l.value;
+        int128_t rValue = Decimal::isWideDecimalType(r.precision)
+            ? r.s128Value : r.value;
+
+        opOverflowCheck(lValue, rValue);
+
+        if (result.scale >= l.scale - r.scale)
+        {
+            int128_t scaleMultiplier;
+
+            getScaleDivisor(scaleMultiplier, result.scale - (l.scale - r.scale));
+
+            // TODO How do we check overflow of (int128_t)((__float128)lValue / rValue * scaleMultiplier) ?
+
+            result.s128Value = (int128_t)(( (lValue > 0 && rValue > 0) || (lValue < 0 && rValue < 0) ?
+                                             (__float128)lValue / rValue * scaleMultiplier + 0.5 :
+                                             (__float128)lValue / rValue * scaleMultiplier - 0.5));
+        }
+        else
+        {
+            int128_t scaleMultiplier;
+
+            getScaleDivisor(scaleMultiplier, (l.scale - r.scale) - result.scale);
+
+            result.s128Value = (int128_t)(( (lValue > 0 && rValue > 0) || (lValue < 0 && rValue < 0) ?
+                                             (__float128)lValue / rValue / scaleMultiplier + 0.5 :
+                                             (__float128)lValue / rValue / scaleMultiplier - 0.5));
+        }
+    }
+
+    template<typename OpOverflowCheck,
+        typename MultiplicationOverflowCheck>
+    void multiplicationExecute(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r,
+        execplan::IDB_Decimal& result,
+        OpOverflowCheck opOverflowCheck,
+        MultiplicationOverflowCheck mulOverflowCheck)
+    {
+        int128_t lValue = Decimal::isWideDecimalType(l.precision)
+            ? l.s128Value : l.value;
+        int128_t rValue = Decimal::isWideDecimalType(r.precision)
+            ? r.s128Value : r.value;
+
+        if (lValue == 0 || rValue == 0)
+        {
+            result.s128Value = 0;
+            return;
+        }
+
+        if (result.scale >= l.scale + r.scale)
+        {
+            int128_t scaleMultiplier;
+
+            getScaleDivisor(scaleMultiplier, result.scale - (l.scale + r.scale));
+
+            opOverflowCheck(lValue, rValue, result.s128Value);
+            opOverflowCheck(result.s128Value, scaleMultiplier, result.s128Value);
+        }
+        else
+        {
+            unsigned int diff = l.scale + r.scale - result.scale;
+
+            int128_t scaleMultiplierL, scaleMultiplierR;
+
+            getScaleDivisor(scaleMultiplierL, diff / 2);
+            getScaleDivisor(scaleMultiplierR, diff - (diff / 2));
+
+            lValue = (int128_t)(( (lValue > 0) ?
+                                   (__float128)lValue / scaleMultiplierL + 0.5 :
+                                   (__float128)lValue / scaleMultiplierL - 0.5));
+
+            rValue = (int128_t)(( (rValue > 0) ?
+                                   (__float128)rValue / scaleMultiplierR + 0.5 :
+                                   (__float128)rValue / scaleMultiplierR - 0.5));
+
+            opOverflowCheck(lValue, rValue, result.s128Value);;
+        }
+    }
+
     std::string Decimal::toString(execplan::IDB_Decimal& value)
     {
         char buf[utils::MAXLENGTH16BYTES];
@@ -163,13 +255,9 @@ namespace datatypes
     {
         std::plus<int128_t> add;
         NoOverflowCheck noOverflowCheck;
-        execute(l, r, result, add, noOverflowCheck, noOverflowCheck);
+        addSubtractExecute(l, r, result, add, noOverflowCheck, noOverflowCheck);
     }
 
-    template
-    void Decimal::addition<int128_t, false>(const execplan::IDB_Decimal& l,
-        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result);
-
     // with overflow check
     template<>
     void Decimal::addition<int128_t, true>(const execplan::IDB_Decimal& l,
@@ -178,13 +266,10 @@ namespace datatypes
         std::plus<int128_t> add;
         AdditionOverflowCheck overflowCheck;
         MultiplicationOverflowCheck mulOverflowCheck;
-        execute(l, r, result, add, overflowCheck, mulOverflowCheck);
+        addSubtractExecute(l, r, result, add, overflowCheck, mulOverflowCheck);
     }
 
-    template
+    // no overflow check
-    void Decimal::addition<int128_t, false>(const execplan::IDB_Decimal& l,
-        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result);
-
     template<>
     void Decimal::addition<int64_t, false>(const execplan::IDB_Decimal& l,
         const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
@@ -195,86 +280,252 @@ namespace datatypes
             return;
         }
 
-        int64_t lValue = 0, rValue = 0;
+        int64_t lValue = l.value, rValue = r.value;
 
-        if (result.scale >= l.scale)
+        if (result.scale > l.scale)
-            lValue = l.value * mcs_pow_10[result.scale - l.scale];
+            lValue *= mcs_pow_10[result.scale - l.scale];
-        else
+        else if (result.scale < l.scale)
-            lValue = (int64_t)(l.value > 0 ?
+            lValue = (int64_t)(lValue > 0 ?
-                                  (double)l.value / mcs_pow_10[l.scale - result.scale] + 0.5 :
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] + 0.5 :
-                                  (double)l.value / mcs_pow_10[l.scale - result.scale] - 0.5);
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] - 0.5);
 
-        if (result.scale >= r.scale)
+        if (result.scale > r.scale)
-            rValue = r.value * mcs_pow_10[result.scale - r.scale];
+            rValue *= mcs_pow_10[result.scale - r.scale];
-        else
+        else if (result.scale < r.scale)
-            rValue = (int64_t)(r.value > 0 ?
+            rValue = (int64_t)(rValue > 0 ?
-                                  (double)r.value / mcs_pow_10[r.scale - result.scale] + 0.5 :
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] + 0.5 :
-                                  (double)r.value / mcs_pow_10[r.scale - result.scale] - 0.5);
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] - 0.5);
 
         result.value = lValue + rValue;
     }
-    template
-    void Decimal::addition<int64_t, false>(const execplan::IDB_Decimal& l,
-        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result);
 
+    // with overflow check
     template<>
     void Decimal::addition<int64_t, true>(const execplan::IDB_Decimal& l,
         const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
     {
-        throw logging::NotImplementedExcept("Decimal::addition<int64>");
+        AdditionOverflowCheck additionOverflowCheck;
+        MultiplicationOverflowCheck mulOverflowCheck;
+
+        if (result.scale == l.scale && result.scale == r.scale)
+        {
+            additionOverflowCheck(l.value, r.value);
+            result.value = l.value + r.value;
+            return;
+        }
+
+        int64_t lValue = l.value, rValue = r.value;
+
+        if (result.scale > l.scale)
+            mulOverflowCheck(lValue, mcs_pow_10[result.scale - l.scale], lValue);
+        else if (result.scale < l.scale)
+            lValue = (int64_t)(lValue > 0 ?
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] + 0.5 :
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] - 0.5);
+
+        if (result.scale > r.scale)
+            mulOverflowCheck(rValue, mcs_pow_10[result.scale - r.scale], rValue);
+        else if (result.scale < r.scale)
+            rValue = (int64_t)(rValue > 0 ?
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] + 0.5 :
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] - 0.5);
+
+        additionOverflowCheck(lValue, rValue);
+        result.value = lValue + rValue;
     }
 
+    // no overflow check
+    template<>
+    void Decimal::subtraction<int128_t, false>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        std::minus<int128_t> subtract;
+        NoOverflowCheck noOverflowCheck;
+        addSubtractExecute(l, r, result, subtract, noOverflowCheck, noOverflowCheck);
+    }
+
+    // with overflow check
+    template<>
+    void Decimal::subtraction<int128_t, true>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        std::minus<int128_t> subtract;
+        SubtractionOverflowCheck overflowCheck;
+        MultiplicationOverflowCheck mulOverflowCheck;
+        addSubtractExecute(l, r, result, subtract, overflowCheck, mulOverflowCheck);
+    }
+
+    // no overflow check
+    template<>
+    void Decimal::subtraction<int64_t, false>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        if (result.scale == l.scale && result.scale == r.scale)
+        {
+            result.value = l.value - r.value;
+            return;
+        }
+
+        int64_t lValue = l.value, rValue = r.value;
+
+        if (result.scale > l.scale)
+            lValue *= mcs_pow_10[result.scale - l.scale];
+        else if (result.scale < l.scale)
+            lValue = (int64_t)(lValue > 0 ?
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] + 0.5 :
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] - 0.5);
+
+        if (result.scale > r.scale)
+            rValue *= mcs_pow_10[result.scale - r.scale];
+        else if (result.scale < r.scale)
+            rValue = (int64_t)(rValue > 0 ?
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] + 0.5 :
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] - 0.5);
+
+        result.value = lValue - rValue;
+    }
+
+    // with overflow check
+    template<>
+    void Decimal::subtraction<int64_t, true>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        SubtractionOverflowCheck subtractionOverflowCheck;
+        MultiplicationOverflowCheck mulOverflowCheck;
+
+        if (result.scale == l.scale && result.scale == r.scale)
+        {
+            subtractionOverflowCheck(l.value, r.value);
+            result.value = l.value - r.value;
+            return;
+        }
+
+        int64_t lValue = l.value, rValue = r.value;
+
+        if (result.scale > l.scale)
+            mulOverflowCheck(lValue, mcs_pow_10[result.scale - l.scale], lValue);
+        else if (result.scale < l.scale)
+            lValue = (int64_t)(lValue > 0 ?
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] + 0.5 :
+                               (double)lValue / mcs_pow_10[l.scale - result.scale] - 0.5);
+
+        if (result.scale > r.scale)
+            mulOverflowCheck(rValue, mcs_pow_10[result.scale - r.scale], rValue);
+        else if (result.scale < r.scale)
+            rValue = (int64_t)(rValue > 0 ?
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] + 0.5 :
+                               (double)rValue / mcs_pow_10[r.scale - result.scale] - 0.5);
+
+        subtractionOverflowCheck(lValue, rValue);
+        result.value = lValue - rValue;
+    }
+
+    // no overflow check
     template<>
     void Decimal::division<int128_t, false>(const execplan::IDB_Decimal& l,
         const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
     {
-        std::divides<int128_t> division;
         NoOverflowCheck noOverflowCheck;
-        execute(l, r, result, division, noOverflowCheck, noOverflowCheck);
+        divisionExecute(l, r, result, noOverflowCheck, noOverflowCheck);
     }
 
-    template
-    void Decimal::division<int128_t, false>(const execplan::IDB_Decimal& l,
-        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result);
-
     // With overflow check
     template<>
     void Decimal::division<int128_t, true>(const execplan::IDB_Decimal& l,
         const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
     {
-        std::divides<int128_t> division;
         DivisionOverflowCheck overflowCheck;
         MultiplicationOverflowCheck mulOverflowCheck;
-        execute(l, r, result, division, overflowCheck, mulOverflowCheck);
+        divisionExecute(l, r, result, overflowCheck, mulOverflowCheck);
     }
 
+    // no overflow check
     // We rely on the zero check from ArithmeticOperator::execute
     template<>
     void Decimal::division<int64_t, false>(const execplan::IDB_Decimal& l,
         const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
     {
         if (result.scale >= l.scale - r.scale)
-            result.value = (int64_t)(( (l.value > 0 && r.value > 0) 
+            result.value = (int64_t)(( (l.value > 0 && r.value > 0) || (l.value < 0 && r.value < 0) ?
-                || (l.value < 0 && r.value < 0) ?
+                                       (long double)l.value / r.value * mcs_pow_10[result.scale - (l.scale - r.scale)] + 0.5 :
-                    (long double)l.value / r.value * mcs_pow_10[result.scale - (l.scale - r.scale)] + 0.5 :
+                                       (long double)l.value / r.value * mcs_pow_10[result.scale - (l.scale - r.scale)] - 0.5));
-                    (long double)l.value / r.value * mcs_pow_10[result.scale - (l.scale - r.scale)] - 0.5));
+        else
-            else
+            result.value = (int64_t)(( (l.value > 0 && r.value > 0) || (l.value < 0 && r.value < 0) ?
-                result.value = (int64_t)(( (l.value > 0 && r.value > 0)
+                                       (long double)l.value / r.value / mcs_pow_10[l.scale - r.scale - result.scale] + 0.5 :
-                    || (l.value < 0 && r.value < 0) ?
+                                       (long double)l.value / r.value / mcs_pow_10[l.scale - r.scale - result.scale] - 0.5));
-                        (long double)l.value / r.value / mcs_pow_10[l.scale - r.scale - result.scale] + 0.5 :
-                        (long double)l.value / r.value / mcs_pow_10[l.scale - r.scale - result.scale] - 0.5));
-
     }
 
-    template
+    // With overflow check
-    void Decimal::division<int64_t, false>(const execplan::IDB_Decimal& l,
-        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result);
-
     template<>
     void Decimal::division<int64_t, true>(const execplan::IDB_Decimal& l,
         const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
     {
-        throw logging::NotImplementedExcept("Decimal::division<int64>");
+        DivisionOverflowCheck divisionOverflowCheck;
+
+        divisionOverflowCheck(l.value, r.value);
+
+        if (result.scale >= l.scale - r.scale)
+            // TODO How do we check overflow of (int64_t)((long double)l.value / r.value * mcs_pow_10[result.scale - (l.scale - r.scale)]) ?
+            result.value = (int64_t)(( (l.value > 0 && r.value > 0) || (l.value < 0 && r.value < 0) ?
+                                       (long double)l.value / r.value * mcs_pow_10[result.scale - (l.scale - r.scale)] + 0.5 :
+                                       (long double)l.value / r.value * mcs_pow_10[result.scale - (l.scale - r.scale)] - 0.5));
+        else
+            result.value = (int64_t)(( (l.value > 0 && r.value > 0) || (l.value < 0 && r.value < 0) ?
+                                       (long double)l.value / r.value / mcs_pow_10[l.scale - r.scale - result.scale] + 0.5 :
+                                       (long double)l.value / r.value / mcs_pow_10[l.scale - r.scale - result.scale] - 0.5));
+    }
+
+    // no overflow check
+    template<>
+    void Decimal::multiplication<int128_t, false>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        MultiplicationNoOverflowCheck noOverflowCheck;
+        multiplicationExecute(l, r, result, noOverflowCheck, noOverflowCheck);
+    }
+
+    // With overflow check
+    template<>
+    void Decimal::multiplication<int128_t, true>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        MultiplicationOverflowCheck mulOverflowCheck;
+        multiplicationExecute(l, r, result, mulOverflowCheck, mulOverflowCheck);
+    }
+
+    // no overflow check
+    template<>
+    void Decimal::multiplication<int64_t, false>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        if (result.scale >= l.scale + r.scale)
+            result.value = l.value * r.value * mcs_pow_10[result.scale - (l.scale + r.scale)];
+        else
+            result.value = (int64_t)(( (l.value > 0 && r.value > 0) || (l.value < 0 && r.value < 0) ?
+                                       (double)l.value * r.value / mcs_pow_10[l.scale + r.scale - result.scale] + 0.5 :
+                                       (double)l.value * r.value / mcs_pow_10[l.scale + r.scale - result.scale] - 0.5));
+    }
+
+    // With overflow check
+    template<>
+    void Decimal::multiplication<int64_t, true>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        MultiplicationOverflowCheck mulOverflowCheck;
+
+        if (result.scale >= l.scale + r.scale)
+        {
+            mulOverflowCheck(l.value, r.value, result.value);
+            mulOverflowCheck(result.value, mcs_pow_10[result.scale - (l.scale + r.scale)], result.value);
+        }
+        else
+        {
+            mulOverflowCheck(l.value, r.value, result.value);
+
+            result.value = (int64_t)(( (result.value > 0) ?
+                                       (double)result.value / mcs_pow_10[l.scale + r.scale - result.scale] + 0.5 :
+                                       (double)result.value / mcs_pow_10[l.scale + r.scale - result.scale] - 0.5));
+        }
     }
 
 } // end of namespace

datatypes/mcs_decimal.h

@@ -35,6 +35,7 @@ namespace datatypes
 constexpr uint32_t MAXDECIMALWIDTH = 16U;
 constexpr uint8_t INT64MAXPRECISION = 18U;
 constexpr uint8_t INT128MAXPRECISION = 38U;
+constexpr uint8_t MAXLEGACYWIDTH = 8U;
 
 const uint64_t mcs_pow_10[20] =
 {
@@ -126,6 +127,15 @@ class Decimal
             const execplan::IDB_Decimal& r,
             execplan::IDB_Decimal& result);
 
+        /**
+            @brief Subtraction template that supports overflow check and
+            two internal representations of decimal.
+        */
+        template<typename T, bool overflow>
+        static void subtraction(const execplan::IDB_Decimal& l,
+            const execplan::IDB_Decimal& r,
+            execplan::IDB_Decimal& result);
+
         /**
             @brief Division template that supports overflow check and
             two internal representations of decimal.
@@ -136,7 +146,16 @@ class Decimal
             execplan::IDB_Decimal& result);
 
         /**
-            @brief Convinience method to put decimal into a std:;string.
+            @brief Multiplication template that supports overflow check and
+            two internal representations of decimal.
+        */
+        template<typename T, bool overflow>
+        static void multiplication(const execplan::IDB_Decimal& l,
+            const execplan::IDB_Decimal& r,
+            execplan::IDB_Decimal& result);
+
+        /**
+            @brief Convenience method to put decimal into a std::string.
         */
         static std::string toString(execplan::IDB_Decimal& value);
 
@@ -161,6 +180,66 @@ class Decimal
                 && precision <= INT128MAXPRECISION;
         }
 
+        /**
+            @brief The method sets the legacy scale and precision of a wide decimal
+            column which is the result of an arithmetic operation.
+        */
+        static inline void setDecimalScalePrecisionLegacy(execplan::CalpontSystemCatalog::ColType& ct,
+            unsigned int precision, unsigned int scale)
+        {
+            ct.scale = scale;
+
+            if (ct.scale == 0)
+                ct.precision = precision - 1;
+            else
+                ct.precision = precision - scale;
+        }
+
+        /**
+            @brief The method sets the scale and precision of a wide decimal
+            column which is the result of an arithmetic operation.
+        */
+        static inline void setDecimalScalePrecision(execplan::CalpontSystemCatalog::ColType& ct,
+            unsigned int precision, unsigned int scale)
+        {
+            ct.colWidth = (precision > INT64MAXPRECISION)
+                ? MAXDECIMALWIDTH : MAXLEGACYWIDTH;
+
+            ct.precision = (precision > INT128MAXPRECISION)
+                ? INT128MAXPRECISION : precision;
+
+            ct.scale = scale;
+        }
+
+        /**
+            @brief The method sets the scale and precision of a wide decimal
+            column which is the result of an arithmetic operation, based on a heuristic.
+        */
+        static inline void setDecimalScalePrecisionHeuristic(execplan::CalpontSystemCatalog::ColType& ct,
+            unsigned int precision, unsigned int scale)
+        {
+            unsigned int diff = 0;
+
+            if (precision > INT128MAXPRECISION)
+            {
+                ct.precision = INT128MAXPRECISION;
+                diff = precision - INT128MAXPRECISION;
+            }
+            else
+            {
+                ct.precision = precision;
+            }
+
+            ct.scale = scale;
+
+            if (diff != 0)
+            {
+                ct.scale = scale - (int)(diff * (38.0/65.0));
+
+                if (ct.scale < 0)
+                    ct.scale = 0;
+            }
+        }
 };
 
 /**
@@ -170,17 +249,25 @@ class Decimal
 struct DivisionOverflowCheck {
     void operator()(const int128_t& x, const int128_t& y)
     {
-        if (x == Decimal::maxInt128 && y == -1)
+        if (x == Decimal::minInt128 && y == -1)
         {
             throw logging::OperationOverflowExcept(
                 "Decimal::division<int128_t> produces an overflow.");
         }
     }
+    void operator()(const int64_t x, const int64_t y)
+    {
+        if (x == std::numeric_limits<int64_t>::min() && y == -1)
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::division<int64_t> produces an overflow.");
+        }
+    }
 };
 
 /**
     @brief The structure contains an overflow check for int128
-    addition.
+    and int64_t multiplication.
 */
 struct MultiplicationOverflowCheck {
     void operator()(const int128_t& x, const int128_t& y)
@@ -189,7 +276,7 @@ struct MultiplicationOverflowCheck {
         {
             throw logging::OperationOverflowExcept(
                 "Decimal::multiplication<int128_t> or scale multiplication \
- produces an overflow.");
+produces an overflow.");
         }
     }
     bool operator()(const int128_t& x, const int128_t& y, int128_t& r)
@@ -198,16 +285,45 @@ struct MultiplicationOverflowCheck {
         {
             throw logging::OperationOverflowExcept(
                 "Decimal::multiplication<int128_t> or scale multiplication \
- produces an overflow.");
+produces an overflow.");
         }
         return true;
     }
+    void operator()(const int64_t x, const int64_t y)
+    {
+        if (x * y / y != x)
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::multiplication<int64_t> or scale multiplication \
+produces an overflow.");
+        }
+    }
+    bool operator()(const int64_t x, const int64_t y, int64_t& r)
+    {
+        if ((r = x * y) / y != x)
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::multiplication<int64_t> or scale multiplication \
+produces an overflow.");
+        }
+        return true;
+    }
+};
 
+/**
+    @brief The strucuture runs an empty overflow check for int128
+    multiplication operation.
+*/
+struct MultiplicationNoOverflowCheck {
+    void operator()(const int128_t& x, const int128_t& y, int128_t& r)
+    {
+        r = x * y;
+    }
 };
 
 /**
     @brief The structure contains an overflow check for int128
-    addition.
+    and int64 addition.
 */
 struct AdditionOverflowCheck {
     void operator()(const int128_t& x, const int128_t& y)
@@ -219,6 +335,40 @@ struct AdditionOverflowCheck {
                 "Decimal::addition<int128_t> produces an overflow.");
         }
     }
+    void operator()(const int64_t x, const int64_t y)
+    {
+        if ((y > 0 && x > std::numeric_limits<int64_t>::max() - y)
+            || (y < 0 && x < std::numeric_limits<int64_t>::min() - y))
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::addition<int64_t> produces an overflow.");
+        }
+    }
+};
+
+/**
+    @brief The structure contains an overflow check for int128
+    subtraction.
+*/
+struct SubtractionOverflowCheck {
+    void operator()(const int128_t& x, const int128_t& y)
+    {
+        if ((y > 0 && x < Decimal::minInt128 + y)
+            || (y < 0 && x > Decimal::maxInt128 + y))
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::subtraction<int128_t> produces an overflow.");
+        }
+    }
+    void operator()(const int64_t x, const int64_t y)
+    {
+        if ((y > 0 && x < std::numeric_limits<int64_t>::min() + y)
+            || (y < 0 && x > std::numeric_limits<int64_t>::max() + y))
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::subtraction<int64_t> produces an overflow.");
+        }
+    }
 };
 
 /**

dbcon/execplan/arithmeticoperator.cpp

@@ -35,19 +35,19 @@ namespace execplan
  * Constructors/Destructors
  */
 ArithmeticOperator::ArithmeticOperator() : Operator(), 
-    fDecimalOverflowCheck(true)
+    fDecimalOverflowCheck(false)
 {
 }
 
 ArithmeticOperator::ArithmeticOperator(const string& operatorName): Operator(operatorName),
-    fDecimalOverflowCheck(true)
+    fDecimalOverflowCheck(false)
 {
 }
 
 ArithmeticOperator::ArithmeticOperator(const ArithmeticOperator& rhs):
     Operator(rhs),
     fTimeZone(rhs.timeZone()),
-    fDecimalOverflowCheck(true)
+    fDecimalOverflowCheck(false)
 {
 }
 

dbcon/execplan/arithmeticoperator.h

@@ -196,7 +196,7 @@ public:
         return TreeNode::getBoolVal();
     }
     void adjustResultType(const CalpontSystemCatalog::ColType& m);
-    constexpr inline bool getOverflowCheck()
+    constexpr inline bool getOverflowCheck() const
     {
         return fDecimalOverflowCheck;
     }
@@ -209,7 +209,6 @@ private:
     template <typename result_t>
     inline result_t execute(result_t op1, result_t op2, bool& isNull);
     inline void execute(IDB_Decimal& result, IDB_Decimal op1, IDB_Decimal op2, bool& isNull);
-    inline void execute(IDB_Decimal& result, IDB_Decimal op1, IDB_Decimal op2, bool& isNull, cscType& resultCscType);
     std::string fTimeZone;
     bool fDecimalOverflowCheck;
 };
@@ -250,7 +249,7 @@ inline void ArithmeticOperator::evaluate(rowgroup::Row& row, bool& isNull, Parse
         // WIP MCOL-641
         case execplan::CalpontSystemCatalog::DECIMAL:
         case execplan::CalpontSystemCatalog::UDECIMAL:
-            execute(fResult.decimalVal, lop->getDecimalVal(row, isNull), rop->getDecimalVal(row, isNull), isNull, fOperationType);
+            execute(fResult.decimalVal, lop->getDecimalVal(row, isNull), rop->getDecimalVal(row, isNull), isNull);
             break;
         default:
         {
@@ -292,20 +291,36 @@ inline result_t ArithmeticOperator::execute(result_t op1, result_t op2, bool& is
     }
 }
 
-inline void ArithmeticOperator::execute(IDB_Decimal& result, IDB_Decimal op1, IDB_Decimal op2, bool& isNull, cscType& resultCscType)
+inline void ArithmeticOperator::execute(IDB_Decimal& result, IDB_Decimal op1, IDB_Decimal op2, bool& isNull)
 {
     switch (fOp)
     {
         case OP_ADD:
-            if (resultCscType.colWidth == datatypes::MAXDECIMALWIDTH)
+            if (fOperationType.colWidth == datatypes::MAXDECIMALWIDTH)
             {
-                datatypes::Decimal::addition<decltype(result.s128Value),true>(
+                if (LIKELY(!fDecimalOverflowCheck))
-                    op1, op2, result);
+                {
+                    datatypes::Decimal::addition<decltype(result.s128Value), false>(
+                        op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::addition<decltype(result.s128Value), true>(
+                        op1, op2, result);
+                }
             }
-            else if (resultCscType.colWidth == utils::MAXLEGACYWIDTH)
+            else if (fOperationType.colWidth == utils::MAXLEGACYWIDTH)
             {
-                datatypes::Decimal::addition<decltype(result.value),false>(
+                if (LIKELY(!fDecimalOverflowCheck))
-                    op1, op2, result);
+                {
+                    datatypes::Decimal::addition<decltype(result.value), false>(
+                        op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::addition<decltype(result.value), true>(
+                        op1, op2, result);
+                }
             }
             else
             {
@@ -315,52 +330,95 @@ inline void ArithmeticOperator::execute(IDB_Decimal& result, IDB_Decimal op1, ID
             break;
 
         case OP_SUB:
-            if (result.scale == op1.scale && result.scale == op2.scale)
+            if (fOperationType.colWidth == datatypes::MAXDECIMALWIDTH)
             {
-                result.value = op1.value - op2.value;
+                if (LIKELY(!fDecimalOverflowCheck))
-                break;
+                {
+                    datatypes::Decimal::subtraction<decltype(result.s128Value), false>(
+                        op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::subtraction<decltype(result.s128Value), true>(
+                        op1, op2, result);
+                }
+            }
+            else if (fOperationType.colWidth == utils::MAXLEGACYWIDTH)
+            {
+                if (LIKELY(!fDecimalOverflowCheck))
+                {
+                    datatypes::Decimal::subtraction<decltype(result.value), false>(
+                        op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::subtraction<decltype(result.value), true>(
+                        op1, op2, result);
+                }
             }
-
-            if (result.scale >= op1.scale)
-                op1.value *= IDB_pow[result.scale - op1.scale];
             else
-                op1.value = (int64_t)(op1.value > 0 ?
+            {
-                                      (double)op1.value / IDB_pow[op1.scale - result.scale] + 0.5 :
+                throw logging::InvalidArgumentExcept(
-                                      (double)op1.value / IDB_pow[op1.scale - result.scale] - 0.5);
+                    "Unexpected result width");
-
+            }
-            if (result.scale >= op2.scale)
-                op2.value *= IDB_pow[result.scale - op2.scale];
-            else
-                op2.value = (int64_t)(op2.value > 0 ?
-                                      (double)op2.value / IDB_pow[op2.scale - result.scale] + 0.5 :
-                                      (double)op2.value / IDB_pow[op2.scale - result.scale] - 0.5);
-
-            result.value = op1.value - op2.value;
             break;
 
         case OP_MUL:
-            if (result.scale >= op1.scale + op2.scale)
+            if (fOperationType.colWidth == datatypes::MAXDECIMALWIDTH)
-                result.value = op1.value * op2.value * IDB_pow[result.scale - (op1.scale + op2.scale)];
+            {
+                if (LIKELY(!fDecimalOverflowCheck))
+                {
+                    datatypes::Decimal::multiplication<decltype(result.s128Value), false>(
+                    op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::multiplication<decltype(result.s128Value), true>(
+                    op1, op2, result);
+                }
+            }
+            else if (fOperationType.colWidth == utils::MAXLEGACYWIDTH)
+            {
+                if (LIKELY(!fDecimalOverflowCheck))
+                {
+                    datatypes::Decimal::multiplication<decltype(result.value), false>(
+                        op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::multiplication<decltype(result.value), true>(
+                        op1, op2, result);
+                }
+            }
             else
-                result.value = (int64_t)(( (op1.value > 0 && op2.value > 0) || (op1.value < 0 && op2.value < 0) ?
+            {
-                                           (double)op1.value * op2.value / IDB_pow[op1.scale + op2.scale - result.scale] + 0.5 :
+                throw logging::InvalidArgumentExcept(
-                                           (double)op1.value * op2.value / IDB_pow[op1.scale + op2.scale - result.scale] - 0.5));
+                    "Unexpected result width");
-
+            }
             break;
 
         case OP_DIV:
-            if (resultCscType.colWidth == 16)
+            if (fOperationType.colWidth == datatypes::MAXDECIMALWIDTH)
             {
-                if (op2.s128Value == 0)
+                if ((datatypes::Decimal::isWideDecimalType(op2.precision) && op2.s128Value == 0)
+                    || (!datatypes::Decimal::isWideDecimalType(op2.precision) && op2.value == 0))
                 {
                     isNull = true;
                     break;
                 }
 
-                datatypes::Decimal::division<decltype(result.s128Value),false>(
+                if (LIKELY(!fDecimalOverflowCheck))
-                op1, op2, result);
+                {
+                    datatypes::Decimal::division<decltype(result.s128Value), false>(
+                    op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::division<decltype(result.s128Value), true>(
+                    op1, op2, result);
+                }
             }
-            else if (resultCscType.colWidth == 8)
+            else if (fOperationType.colWidth == utils::MAXLEGACYWIDTH)
             {
                 if (op2.value == 0)
                 {
@@ -368,8 +426,16 @@ inline void ArithmeticOperator::execute(IDB_Decimal& result, IDB_Decimal op1, ID
                     break;
                 }
 
-                datatypes::Decimal::division<decltype(result.value),false>(
+                if (LIKELY(!fDecimalOverflowCheck))
-                    op1, op2, result);
+                {
+                    datatypes::Decimal::division<decltype(result.value), false>(
+                        op1, op2, result);
+                }
+                else
+                {
+                    datatypes::Decimal::division<decltype(result.value), true>(
+                        op1, op2, result);
+                }
             }
             else
             {

dbcon/mysql/ha_mcs_execplan.cpp

@@ -3105,17 +3105,13 @@ CalpontSystemCatalog::ColType colType_MysqlToIDB (const Item* item)
         case DECIMAL_RESULT:
         {
             Item_decimal* idp = (Item_decimal*)item;
+
             ct.colDataType = CalpontSystemCatalog::DECIMAL;
-            ct.colWidth = (idp->max_length >= datatypes::INT64MAXPRECISION)
+
-                ? datatypes::MAXDECIMALWIDTH : utils::MAXLEGACYWIDTH;
+            unsigned int precision = idp->decimal_precision();
-            ct.scale = idp->decimals;
+            unsigned int scale = idp->decimal_scale();
-            // WIP MCOL-641
+
-            if (ct.scale == 0)
+            datatypes::Decimal::setDecimalScalePrecision(ct, precision, scale);
-                ct.precision = (idp->max_length > datatypes::INT128MAXPRECISION)
-                    ? datatypes::INT128MAXPRECISION : idp->max_length - 1;
-            else
-                ct.precision = (idp->max_length > datatypes::INT128MAXPRECISION )
-                    ? datatypes::INT128MAXPRECISION : idp->max_length - idp->decimals;
 
             break;
         }
@@ -3141,7 +3137,7 @@ ReturnedColumn* buildReturnedColumn(
 {
     ReturnedColumn* rc = NULL;
 
-    if ( gwi.thd)
+    if (gwi.thd)
     {
         //if ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) || ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ))
         {
@@ -3444,6 +3440,7 @@ ArithmeticColumn* buildArithmeticColumn(
     Item** sfitempp = item->arguments();
     ArithmeticOperator* aop = new ArithmeticOperator(item->func_name());
     aop->timeZone(gwi.thd->variables.time_zone->get_name()->ptr());
+    aop->setOverflowCheck(get_decimal_overflow_check(gwi.thd));
     ParseTree* pt = new ParseTree(aop);
     //ReturnedColumn *lhs = 0, *rhs = 0;
     ParseTree* lhs = 0, *rhs = 0;
@@ -3606,6 +3603,60 @@ ArithmeticColumn* buildArithmeticColumn(
     // decimal arithmetic operation gives double result when the session variable is set.
     CalpontSystemCatalog::ColType mysql_type = colType_MysqlToIDB(item);
 
+    if (mysql_type.colDataType == CalpontSystemCatalog::DECIMAL ||
+        mysql_type.colDataType == CalpontSystemCatalog::UDECIMAL)
+    {
+        int32_t leftColWidth = pt->left()->data()->resultType().colWidth;
+        int32_t rightColWidth = pt->right()->data()->resultType().colWidth;
+
+        // Revert back to legacy values of scale and precision
+        // if the 2 columns involved in the expression are not wide
+        if (leftColWidth <= utils::MAXLEGACYWIDTH &&
+            rightColWidth <= utils::MAXLEGACYWIDTH)
+        {
+            Item_decimal* idp = (Item_decimal*)item;
+
+            mysql_type.colWidth = 8;
+
+            unsigned int precision = idp->max_length;
+            unsigned int scale = idp->decimals;
+
+            datatypes::Decimal::setDecimalScalePrecisionLegacy(mysql_type, precision, scale);
+        }
+        else
+        {
+
+            if (leftColWidth == datatypes::MAXDECIMALWIDTH ||
+                rightColWidth == datatypes::MAXDECIMALWIDTH)
+                mysql_type.colWidth = datatypes::MAXDECIMALWIDTH;
+
+            if (mysql_type.colWidth == datatypes::MAXDECIMALWIDTH)
+            {
+                string funcName = item->func_name();
+
+                int32_t scale1 = pt->left()->data()->resultType().scale;
+                int32_t scale2 = pt->right()->data()->resultType().scale;
+
+                if (funcName == "/" &&
+                    (mysql_type.scale - (scale1 - scale2)) > datatypes::INT128MAXPRECISION)
+                {
+                    Item_decimal* idp = (Item_decimal*)item;
+
+                    unsigned int precision = idp->decimal_precision();
+                    unsigned int scale = idp->decimal_scale();
+
+                    datatypes::Decimal::setDecimalScalePrecisionHeuristic(mysql_type, precision, scale);
+
+                    if (mysql_type.scale < scale1)
+                        mysql_type.scale = scale1;
+
+                    if (mysql_type.precision < mysql_type.scale)
+                        mysql_type.precision = mysql_type.scale;
+                }
+            }
+        }
+    }
+
     if (get_double_for_decimal_math(current_thd) == true)
         aop->adjustResultType(mysql_type);
     else

dbcon/mysql/ha_mcs_sysvars.cpp

@@ -159,6 +159,15 @@ static MYSQL_THDVAR_BOOL(
     0
 );
 
+static MYSQL_THDVAR_BOOL(
+    decimal_overflow_check,
+    PLUGIN_VAR_NOCMDARG,
+    "Enable/disable for ColumnStore to check for overflow in arithmetic operation.",
+    NULL,
+    NULL,
+    0
+);
+
 static MYSQL_THDVAR_BOOL(
     ordered_only,
     PLUGIN_VAR_NOCMDARG,
@@ -353,6 +362,7 @@ st_mysql_sys_var* mcs_system_variables[] =
   MYSQL_SYSVAR(diskjoin_bucketsize),
   MYSQL_SYSVAR(um_mem_limit),
   MYSQL_SYSVAR(double_for_decimal_math),
+  MYSQL_SYSVAR(decimal_overflow_check),
   MYSQL_SYSVAR(local_query),
   MYSQL_SYSVAR(use_import_for_batchinsert),
   MYSQL_SYSVAR(import_for_batchinsert_delimiter),
@@ -557,6 +567,15 @@ void set_double_for_decimal_math(THD* thd, bool value)
     THDVAR(thd, double_for_decimal_math) = value;
 }
 
+bool get_decimal_overflow_check(THD* thd)
+{
+    return ( thd == NULL ) ? false : THDVAR(thd, decimal_overflow_check);
+}
+void set_decimal_overflow_check(THD* thd, bool value)
+{
+    THDVAR(thd, decimal_overflow_check) = value;
+}
+
 ulong get_local_query(THD* thd)
 {
     return ( thd == NULL ) ? 0 : THDVAR(thd, local_query);

dbcon/mysql/ha_mcs_sysvars.h

@@ -101,6 +101,9 @@ void set_varbin_always_hex(THD* thd, bool value);
 bool get_double_for_decimal_math(THD* thd);
 void set_double_for_decimal_math(THD* thd, bool value);
 
+bool get_decimal_overflow_check(THD* thd);
+void set_decimal_overflow_check(THD* thd, bool value);
+
 ulong get_local_query(THD* thd);
 void set_local_query(THD* thd, ulong value);