MCOL-641 Initial version of Math operations for wide decimal.

CMakeLists.txt

@@ -200,7 +200,8 @@ ENDIF()
 
 
 SET (ENGINE_LDFLAGS    "-Wl,--no-as-needed -Wl,--add-needed")
-SET (ENGINE_COMMON_LIBS     messageqcpp loggingcpp configcpp idbboot ${Boost_LIBRARIES} xml2 pthread rt libmysql_client datatypes)
+SET (ENGINE_DT_LIB datatypes)
+SET (ENGINE_COMMON_LIBS     messageqcpp loggingcpp configcpp idbboot ${Boost_LIBRARIES} xml2 pthread rt libmysql_client ${ENGINE_DT_LIB})
 SET (ENGINE_OAM_LIBS        oamcpp alarmmanager)
 SET (ENGINE_BRM_LIBS        brm idbdatafile cacheutils rwlock ${ENGINE_OAM_LIBS} ${ENGINE_COMMON_LIBS})
 SET (ENGINE_EXEC_LIBS       joblist execplan windowfunction joiner rowgroup funcexp udfsdk regr dataconvert common compress querystats querytele thrift threadpool ${ENGINE_BRM_LIBS})

datatypes/CMakeLists.txt

@@ -1,7 +1,7 @@
 
 include_directories( ${ENGINE_COMMON_INCLUDES} )
 set(datatypes_LIB_SRCS
-    csdecimal.cpp)
+    mcs_decimal.cpp)
 
 add_library(datatypes SHARED ${datatypes_LIB_SRCS})
 

datatypes/csdecimal.cpp

@@ -1,30 +0,0 @@
-/* Copyright (C) 2020 MariaDB Corporation
-
-   This program is free software; you can redistribute it and/or
-   modify it under the terms of the GNU General Public License
-   as published by the Free Software Foundation; version 2 of
-   the License.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
-   MA 02110-1301, USA. */
-
-#include "csdecimal.h"
-
-const datatypes::Decimal someDecimal;
-
-namespace datatypes
-{
-
-    int Decimal::compare(const execplan::IDB_Decimal& l, const execplan::IDB_Decimal& r)
-    {
-        return 0;
-    }
-
-} // end of namespace

datatypes/csdecimal.h

@@ -1,39 +0,0 @@
-/* Copyright (C) 2020 MariaDB Corporation
-
-   This program is free software; you can redistribute it and/or
-   modify it under the terms of the GNU General Public License
-   as published by the Free Software Foundation; version 2 of
-   the License.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
-   MA 02110-1301, USA. */
-
-#ifndef H_DECIMALDATATYPE
-#define H_DECIMALDATATYPE
-
-#include <cstdint>
-
-namespace execplan
-{
-    struct IDB_Decimal;
-}
-
-namespace datatypes
-{
-class Decimal
-{
-    public:
-        Decimal() { };
-        ~Decimal() { };
-        static int compare(const execplan::IDB_Decimal& l, const execplan::IDB_Decimal& r); 
-};
-
-} //end of namespace
-#endif

datatypes/mcs_decimal.cpp

@@ -0,0 +1,261 @@
+/* Copyright (C) 2020 MariaDB Corporation
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License
+   as published by the Free Software Foundation; version 2 of
+   the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+   MA 02110-1301, USA. */
+
+#include <functional>
+#include <string>
+
+#include "mcs_decimal.h"
+#include "treenode.h"
+#include "exceptclasses.h"
+
+namespace datatypes
+{
+    template<typename BinaryOperation,
+        typename OpOverflowCheck,
+        typename MultiplicationOverflowCheck>
+    void execute(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r,
+        execplan::IDB_Decimal& result,
+        BinaryOperation op,
+        OpOverflowCheck opOverflowCheck,
+        MultiplicationOverflowCheck mulOverflowCheck)
+    {
+        int128_t lValue = Decimal::isWideDecimalType(l.precision)
+            ? l.s128Value : l.value;
+        int128_t rValue = Decimal::isWideDecimalType(l.precision)
+            ? r.s128Value : r.value;
+
+        if (result.scale == l.scale && result.scale == r.scale)
+        {
+            opOverflowCheck(lValue, rValue);
+            result.s128Value = op(lValue, rValue);
+            return;
+        }
+
+        if (result.scale > l.scale)
+        {
+            int128_t scaleMultiplier;
+            getScaleDivisor(scaleMultiplier, result.scale - l.scale);
+            mulOverflowCheck(lValue, scaleMultiplier);
+            lValue *= scaleMultiplier;
+        }
+        else if (result.scale < l.scale)
+        {
+            int128_t scaleMultiplier;
+            getScaleDivisor(scaleMultiplier, l.scale - result.scale);
+            lValue /= scaleMultiplier;
+        }
+
+        if (result.scale > r.scale)
+        {
+            int128_t scaleMultiplier;
+            getScaleDivisor(scaleMultiplier, result.scale - r.scale);
+            mulOverflowCheck(rValue, scaleMultiplier);
+            rValue *= scaleMultiplier;
+        }
+        else if (result.scale < r.scale)
+        {
+            int128_t scaleMultiplier;
+            getScaleDivisor(scaleMultiplier, r.scale - result.scale);
+            mulOverflowCheck(rValue, scaleMultiplier);
+            rValue /= scaleMultiplier;
+        }
+
+        // We assume there is no way that lValue or rValue calculations
+        // give an overflow and this is an incorrect assumption.
+        opOverflowCheck(lValue, rValue);
+
+        result.s128Value = op(lValue, rValue);
+    }
+
+    // This is wide Decimal version only ATM
+    std::string Decimal::toString(execplan::IDB_Decimal& value)
+    {
+        char buf[utils::MAXLENGTH16BYTES];
+        dataconvert::DataConvert::decimalToString(&value.s128Value,
+            value.scale, buf, sizeof(buf),
+            execplan::CalpontSystemCatalog::DECIMAL);
+        return std::string(buf);
+    }
+
+    int Decimal::compare(const execplan::IDB_Decimal& l, const execplan::IDB_Decimal& r)
+    {
+        int128_t divL, divR;
+        getScaleDivisor(divL, l.scale);
+        getScaleDivisor(divR, r.scale);
+        int128_t quotinentL, quotinentR, remainderL, remainderR;
+        quotinentL = l.s128Value/divL;
+        remainderL = l.s128Value%divL;
+        quotinentR = r.s128Value/divR;
+        remainderR = r.s128Value%divR;
+
+        int ret = 0;
+
+        if (quotinentL > quotinentR)
+        {
+            ret = 1;
+        }
+        else if (quotinentL < quotinentR)
+        {
+            ret = -1;
+        }
+        else
+        {
+            // rem carries the value's sign, but needs to be normalized.
+            int32_t s = l.scale - r.scale;
+
+            if (s < 0)
+            {
+                if ((remainderL * mcs_pow_10[-s]) > remainderR)
+                    ret = 1;
+                else if ((remainderL * mcs_pow_10[-s]) < remainderR)
+                    ret = -1;
+            }
+            else
+            {
+                if (remainderL > (remainderR * mcs_pow_10[s]))
+                    ret = 1;
+                else if (remainderL < (remainderR * mcs_pow_10[s]))
+                    ret = -1;
+            }
+        }
+
+        return ret;
+    }
+
+    // no overflow check
+    template<>
+    void Decimal::addition<int128_t, false>(const execplan::IDB_Decimal& l,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        std::plus<int128_t> add;
+        NoOverflowCheck noOverflowCheck;
+        execute(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,
+        const execplan::IDB_Decimal& r, execplan::IDB_Decimal& result)
+    {
+        std::plus<int128_t> add;
+        AdditionOverflowCheck overflowCheck;
+        MultiplicationOverflowCheck mulOverflowCheck;
+        execute(l, r, result, add, overflowCheck, mulOverflowCheck);
+    }
+
+    template
+    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)
+    {
+        if (result.scale == l.scale && result.scale == r.scale)
+        {
+            result.value = l.value + r.value;
+            return;
+        }
+
+        int64_t lValue = 0, rValue = 0;
+
+        if (result.scale >= l.scale)
+            lValue = l.value * mcs_pow_10[result.scale - l.scale];
+        else
+            lValue = (int64_t)(l.value > 0 ?
+                                  (double)l.value / mcs_pow_10[l.scale - result.scale] + 0.5 :
+                                  (double)l.value / mcs_pow_10[l.scale - result.scale] - 0.5);
+
+        if (result.scale >= r.scale)
+            rValue = r.value * mcs_pow_10[result.scale - r.scale];
+        else
+            rValue = (int64_t)(r.value > 0 ?
+                                  (double)r.value / mcs_pow_10[r.scale - result.scale] + 0.5 :
+                                  (double)r.value / 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);
+
+    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>");
+    }
+
+    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);
+    }
+
+    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);
+    }
+
+    // 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) 
+                || (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));
+
+    }
+
+    template
+    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>");
+    }
+
+} // end of namespace

datatypes/mcs_decimal.h

@@ -0,0 +1,224 @@
+/* Copyright (C) 2020 MariaDB Corporation
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License
+   as published by the Free Software Foundation; version 2 of
+   the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+   MA 02110-1301, USA. */
+
+#ifndef H_DECIMALDATATYPE
+#define H_DECIMALDATATYPE
+
+#include <cstdint>
+
+#include "calpontsystemcatalog.h"
+
+using int128_t = __int128;
+
+namespace execplan
+{
+    struct IDB_Decimal;
+}
+
+namespace datatypes
+{
+
+constexpr uint32_t MAXDECIMALWIDTH = 16U;
+constexpr uint8_t INT64MAXPRECISION = 18U;
+constexpr uint8_t INT128MAXPRECISION = 38U;
+
+const uint64_t mcs_pow_10[20] =
+{
+    1ULL,
+    10ULL,
+    100ULL,
+    1000ULL,
+    10000ULL,
+    100000ULL,
+    1000000ULL,
+    10000000ULL,
+    100000000ULL,
+    1000000000ULL,
+    10000000000ULL,
+    100000000000ULL,
+    1000000000000ULL,
+    10000000000000ULL,
+    100000000000000ULL,
+    1000000000000000ULL,
+    10000000000000000ULL,
+    100000000000000000ULL,
+    1000000000000000000ULL,
+    10000000000000000000ULL
+};
+
+constexpr uint32_t maxPowOf10 = sizeof(mcs_pow_10)/sizeof(mcs_pow_10[0])-1;
+constexpr int128_t Decimal128Null = int128_t(0x8000000000000000LL) << 64;
+
+/**
+    @brief The function to produce scale multiplier/divisor for
+    wide decimals.
+*/
+inline void getScaleDivisor(int128_t& divisor, const int8_t scale)
+{
+    divisor = 1;
+    switch (scale/maxPowOf10)
+    {
+        case 2:
+            divisor *= mcs_pow_10[maxPowOf10];
+            divisor *= mcs_pow_10[maxPowOf10];
+            break;
+        case 1:
+            divisor *= mcs_pow_10[maxPowOf10];
+        case 0:
+            divisor *= mcs_pow_10[scale%maxPowOf10];
+        default:
+            break;
+    }
+}
+
+/**
+    @brief The template to generalise common math operation
+    execution path using struct from <functional>.
+*/
+template<typename BinaryOperation, typename OverflowCheck>
+void execute(const execplan::IDB_Decimal& l,
+    const execplan::IDB_Decimal& r,
+    execplan::IDB_Decimal& result,
+    BinaryOperation op,
+    OverflowCheck overflowCheck);
+
+/**
+    @brief Contains subset of decimal related operations.
+
+    Most of the methods are static to allow to call them from
+    the existing code.
+    The main purpose of the class is to collect methods for a
+    base Datatype class.
+*/
+class Decimal
+{
+    public:
+        Decimal() { };
+        ~Decimal() { };
+
+        static constexpr int128_t minInt128 = int128_t(0x8000000000000000LL) << 64;
+        static constexpr int128_t maxInt128 = (int128_t(0x7FFFFFFFFFFFFFFFLL) << 64) + 0xFFFFFFFFFFFFFFFFLL;
+
+        /**
+            @brief Compares two IDB_Decimal taking scale into account. 
+        */
+        static int compare(const execplan::IDB_Decimal& l, const execplan::IDB_Decimal& r); 
+        /**
+            @brief Addition template that supports overflow check and
+            two internal representations of decimal.
+        */
+        template<typename T, bool overflow>
+        static void addition(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.
+        */
+        template<typename T, bool overflow>
+        static void division(const execplan::IDB_Decimal& l,
+            const execplan::IDB_Decimal& r,
+            execplan::IDB_Decimal& result);
+
+        /**
+            @brief Convinience method to put decimal into a std:;string.
+        */
+        static std::string toString(execplan::IDB_Decimal& value);
+
+        /**
+            @brief The method detects whether decimal type is wide
+            using csc data type.
+        */
+        static constexpr inline bool isWideDecimalType(const execplan::CalpontSystemCatalog::ColType& ct)
+        {
+            return ((ct.colDataType == execplan::CalpontSystemCatalog::DECIMAL ||
+                ct.colDataType == execplan::CalpontSystemCatalog::UDECIMAL) &&
+                ct.colWidth == MAXDECIMALWIDTH);
+        }
+
+        /**
+            @brief The method detects whether decimal type is wide
+            using precision.
+        */
+        static constexpr inline bool isWideDecimalType(const int32_t precision)
+        {
+            return precision > INT64MAXPRECISION
+                && precision <= INT128MAXPRECISION;
+        }
+
+};
+
+/**
+    @brief The structure contains an overflow check for int128
+    division.
+*/
+struct DivisionOverflowCheck {
+    void operator()(const int128_t& x, const int128_t& y)
+    {
+        if (x == Decimal::maxInt128 && y == -1)
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::division<int128_t> produces an overflow.");
+        }
+    }
+};
+
+/**
+    @brief The structure contains an overflow check for int128
+    addition.
+*/
+struct MultiplicationOverflowCheck {
+    void operator()(const int128_t& x, const int128_t& y)
+    {
+        if (x * y / x != y)
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::multiplication<int128_t> or scale multiplicationproduces an overflow.");
+        }
+    }
+};
+
+/**
+    @brief The structure contains an overflow check for int128
+    addition.
+*/
+struct AdditionOverflowCheck {
+    void operator()(const int128_t& x, const int128_t& y)
+    {
+        if ((y > 0 && x > Decimal::maxInt128 - y)
+            || (y < 0 && x < Decimal::minInt128 - y))
+        {
+            throw logging::OperationOverflowExcept(
+                "Decimal::addition<int128_t> produces an overflow.");
+        }
+    }
+};
+
+/**
+    @brief The strucuture runs an empty overflow check for int128
+    operation.
+*/
+struct NoOverflowCheck {
+    void operator()(const int128_t& x, const int128_t& y)
+    {
+        return;
+    }
+};
+
+} //end of namespace
+#endif

dbcon/execplan/CMakeLists.txt

@@ -48,7 +48,6 @@ add_library(execplan SHARED ${execplan_LIB_SRCS})
 
 add_dependencies(execplan loggingcpp)
 
-target_link_libraries(execplan ${NETSNMP_LIBRARIES} ${MARIADB_STRING_LIBS})
+target_link_libraries(execplan ${NETSNMP_LIBRARIES} ${MARIADB_STRING_LIBS} ${ENGINE_DT_LIB})
 
 install(TARGETS execplan DESTINATION ${ENGINE_LIBDIR} COMPONENT columnstore-engine)
-

dbcon/execplan/arithmeticoperator.h

@@ -290,31 +290,19 @@ inline void ArithmeticOperator::execute(IDB_Decimal& result, IDB_Decimal op1, ID
         case OP_ADD:
             if (resultCscType.colWidth == 16)
             {
-                result.s128Value = op1.s128Value + op2.s128Value;
+                datatypes::Decimal::addition<decltype(result.s128Value),false>(
-                break;
+                    op1, op2, result);
             }
-
+            else if (resultCscType.colWidth == 8)
-            if (result.scale == op1.scale && result.scale == op2.scale)
             {
-                result.value = op1.value + op2.value;
+                datatypes::Decimal::addition<decltype(result.value),false>(
-                break;
+                    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_SUB:
@@ -352,21 +340,33 @@ inline void ArithmeticOperator::execute(IDB_Decimal& result, IDB_Decimal op1, ID
             break;
 
         case OP_DIV:
-            if (op2.value == 0)
+            if (resultCscType.colWidth == 16)
             {
-                isNull = true;
+                if (op2.s128Value == 0)
-                break;
+                {
+                    isNull = true;
+                    break;
+                }
+
+                datatypes::Decimal::division<decltype(result.s128Value),false>(
+                op1, op2, result);
             }
+            else if (resultCscType.colWidth == 8)
+            {
+                if (op2.value == 0)
+                {
+                    isNull = true;
+                    break;
+                }
 
-            if (result.scale >= op1.scale - op2.scale)
+                datatypes::Decimal::division<decltype(result.value),false>(
-                result.value = (int64_t)(( (op1.value > 0 && op2.value > 0) || (op1.value < 0 && op2.value < 0) ?
+                    op1, op2, result);
-                                           (long double)op1.value / op2.value * IDB_pow[result.scale - (op1.scale - op2.scale)] + 0.5 :
+            }
-                                           (long double)op1.value / op2.value * IDB_pow[result.scale - (op1.scale - op2.scale)] - 0.5));
             else
-                result.value = (int64_t)(( (op1.value > 0 && op2.value > 0) || (op1.value < 0 && op2.value < 0) ?
+            {
-                                           (long double)op1.value / op2.value / IDB_pow[op1.scale - op2.scale - result.scale] + 0.5 :
+                throw logging::InvalidArgumentExcept(
-                                           (long double)op1.value / op2.value / IDB_pow[op1.scale - op2.scale - result.scale] - 0.5));
+                    "Unexpected result width");
-
+            }
             break;
 
         default:

dbcon/execplan/treenode.h

@@ -36,9 +36,7 @@
 #include "exceptclasses.h"
 #include "dataconvert.h"
 #include "columnwidth.h"
-#include "csdecimal.h"
+#include "mcs_decimal.h"
-
-using int128_t = __int128;
 
 namespace messageqcpp
 {

dbcon/joblist/jlf_common.cpp

@@ -38,6 +38,7 @@ using namespace BRM;
 
 #include "jlf_common.h"
 using namespace joblist;
+#include "mcs_decimal.h"
 
 namespace
 {
@@ -332,7 +333,7 @@ string extractTableAlias(const SSC& sc)
 //------------------------------------------------------------------------------
 CalpontSystemCatalog::OID isDictCol(const CalpontSystemCatalog::ColType& colType)
 {
-    if (utils::isWideDecimalType(colType)) return 0;
+    if (datatypes::Decimal::isWideDecimalType(colType)) return 0;
     
     if (colType.colWidth > 8) return colType.ddn.dictOID;
 

dbcon/joblist/jlf_execplantojoblist.cpp

@@ -88,7 +88,7 @@ using namespace logging;
 #include "jlf_common.h"
 #include "jlf_subquery.h"
 #include "jlf_tuplejoblist.h"
-#include "columnwidth.h"
+#include "mcs_decimal.h"
 
 namespace
 {
@@ -1897,7 +1897,7 @@ const JobStepVector doSimpleFilter(SimpleFilter* sf, JobInfo& jobInfo)
             // WIP MCOL-641 width check must be a f() not a literal
             // make a template from convertValueNum to avoid extra if
             // this condition doesn't support UDECIMAL
-            if (utils::isWideDecimalType(ct))
+            if (datatypes::Decimal::isWideDecimalType(ct))
                 convertValueNum(constval, ct, isNull, rf, jobInfo.timeZone, value128);
             else
                 convertValueNum(constval, ct, isNull, rf, jobInfo.timeZone, value);
@@ -1934,7 +1934,7 @@ const JobStepVector doSimpleFilter(SimpleFilter* sf, JobInfo& jobInfo)
 
                 if (sc->isColumnStore())
                 {
-                    if (utils::isWideDecimalType(ct))
+                    if (datatypes::Decimal::isWideDecimalType(ct))
                         pcs->addFilter(cop, value128, rf);
                     else
                         pcs->addFilter(cop, value, rf);
@@ -3012,7 +3012,7 @@ const JobStepVector doConstantFilter(const ConstantFilter* cf, JobInfo& jobInfo)
                     uint8_t rf = 0;
                     bool isNull = ConstantColumn::NULLDATA == cc->type();
 
-                    if (utils::isWideDecimalType(ct))
+                    if (datatypes::Decimal::isWideDecimalType(ct))
                         convertValueNum(constval, ct, isNull, rf, jobInfo.timeZone, value128);
                     else
                         convertValueNum(constval, ct, isNull, rf, jobInfo.timeZone, value);
@@ -3032,7 +3032,7 @@ const JobStepVector doConstantFilter(const ConstantFilter* cf, JobInfo& jobInfo)
                     if (ConstantColumn::NULLDATA == cc->type() && (opeq == *sop || opne == *sop))
                         cop = COMPARE_NIL;
 
-                    if (utils::isWideDecimalType(ct))
+                    if (datatypes::Decimal::isWideDecimalType(ct))
                         pcs->addFilter(cop, value128, rf);
                     else
                         pcs->addFilter(cop, value, rf);

dbcon/joblist/lbidlist.cpp

@@ -28,7 +28,7 @@
 #include "brm.h"
 #include "brmtypes.h"
 #include "dataconvert.h"
-#include "columnwidth.h"
+#include "mcs_decimal.h"
 
 #define IS_VERBOSE (fDebug >= 4)
 #define IS_DETAIL  (fDebug >= 3)
@@ -809,7 +809,7 @@ bool LBIDList::CasualPartitionPredicate(const BRM::EMCasualPartition_t& cpRange,
 
         // Should we also check for empty here?
         // TODO MCOL-641
-        if (utils::isWideDecimalType(ct))
+        if (datatypes::Decimal::isWideDecimalType(ct))
         {
             if (isNull(bigValue, ct))
                 continue;

primitives/primproc/filtercommand.cpp

@@ -27,6 +27,7 @@
 #include "dictstep.h"
 #include "filtercommand.h"
 #include "dataconvert.h"
+#include "mcs_decimal.h"
 
 using namespace std;
 using namespace messageqcpp;
@@ -249,7 +250,7 @@ void FilterCommand::setColTypes(const execplan::CalpontSystemCatalog::ColType& l
     leftColType = left;
     rightColType = right;
 
-    if (utils::isWideDecimalType(left) || utils::isWideDecimalType(right))
+    if (datatypes::Decimal::isWideDecimalType(left) || datatypes::Decimal::isWideDecimalType(right))
         hasWideDecimalType = true;
 }
 
@@ -281,7 +282,7 @@ void FilterCommand::doFilter()
                 bpp->relRids[bpp->ridCount] = bpp->fFiltCmdRids[0][i];
                 // WIP MCOL-641 How is bpp->(binary)values used given that
                 // we are setting the relRids?
-                if (utils::isWideDecimalType(leftColType))
+                if (datatypes::Decimal::isWideDecimalType(leftColType))
                     bpp->binaryValues[bpp->ridCount] = bpp->fFiltCmdBinaryValues[0][i];
                 else
                     bpp->values[bpp->ridCount] = bpp->fFiltCmdValues[0][i];
@@ -343,7 +344,7 @@ bool FilterCommand::binaryCompare(uint64_t i, uint64_t j)
     // not int128_t
     int128_t leftVal, rightVal;
 
-    if (utils::isWideDecimalType(leftColType))
+    if (datatypes::Decimal::isWideDecimalType(leftColType))
     {
         if (execplan::isNull(bpp->fFiltCmdBinaryValues[0][i], leftColType))
             return false;
@@ -356,7 +357,7 @@ bool FilterCommand::binaryCompare(uint64_t i, uint64_t j)
         leftVal = bpp->fFiltCmdValues[0][i];
     }
 
-    if (utils::isWideDecimalType(rightColType))
+    if (datatypes::Decimal::isWideDecimalType(rightColType))
     {
         if (execplan::isNull(bpp->fFiltCmdBinaryValues[1][j], rightColType))
             return false;

tests/CMakeLists.txt

@@ -12,6 +12,12 @@ if (WITH_ARITHMETICOPERATOR_UT)
     install(TARGETS rowgroup_tests DESTINATION ${ENGINE_BINDIR} COMPONENT columnstore-platform)
 endif()
 
+if (WITH_CSDECIMAL_UT)
+    add_executable(mcs_decimal_tests mcs_decimal-tests.cpp)
+    target_link_libraries(mcs_decimal_tests ${ENGINE_LDFLAGS} ${GTEST_LIBRARIES} ${ENGINE_EXEC_LIBS} ${MARIADB_CLIENT_LIBS}) 
+    install(TARGETS rowgroup_tests DESTINATION ${ENGINE_BINDIR} COMPONENT columnstore-platform)
+endif()
+
 if (WITH_DATACONVERT_UT)
     add_executable(dataconvert_tests dataconvert-tests.cpp)
     target_link_libraries(dataconvert_tests ${ENGINE_LDFLAGS} ${GTEST_LIBRARIES} ${ENGINE_EXEC_LIBS} ${MARIADB_CLIENT_LIBS}) 

tests/csdecimal.cpp

@@ -1,23 +0,0 @@
-/* Copyright (C) 2020 MariaDB Corporation
-
-   This program is free software; you can redistribute it and/or
-   modify it under the terms of the GNU General Public License
-   as published by the Free Software Foundation; version 2 of
-   the License.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
-   MA 02110-1301, USA. */
-
-#include "gtest/gtest.h"
-
-TEST(DataConvertTest, Strtoll128)
-{
-    EXPECT_TRUE(true);
-}

tests/mcs_decimal-tests.cpp

@@ -0,0 +1,731 @@
+/* Copyright (C) 2020 MariaDB Corporation
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License
+   as published by the Free Software Foundation; version 2 of
+   the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+   MA 02110-1301, USA. */
+
+#include "gtest/gtest.h"
+
+#include "treenode.h"
+#include "mcs_decimal.h"
+#include "widedecimalutils.h"
+
+TEST(Decimal, compareCheck)
+{
+    // L values = R value, L scale < R scale
+    execplan::IDB_Decimal l, r;
+    l.scale = 20;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 21;
+    l.precision = 38;
+    r.s128Value = 420;
+    EXPECT_EQ(0, datatypes::Decimal::compare(l, r));
+    // L values = R value, L scale > R scale
+    l.scale = 21;
+    l.precision = 38;
+    l.s128Value = 420;
+
+    r.scale = 20;
+    l.precision = 38;
+    r.s128Value = 42;
+    EXPECT_EQ(0, datatypes::Decimal::compare(l, r));
+    // L values > R value, L scale < R scale
+    l.scale = 20;
+    l.precision = 38;
+    l.s128Value = 999999;
+
+    r.scale = 21;
+    l.precision = 38;
+    r.s128Value = 420;
+    EXPECT_EQ(1, datatypes::Decimal::compare(l, r));
+    // L values > R value, L scale > R scale
+    l.scale = 21;
+    l.precision = 38;
+    l.s128Value = 99999999;
+
+    r.scale = 20;
+    l.precision = 38;
+    r.s128Value = 420;
+    EXPECT_EQ(1, datatypes::Decimal::compare(l, r));
+    // L values < R value, L scale < R scale
+    l.scale = 20;
+    l.precision = 38;
+    l.s128Value = 99;
+
+    r.scale = 21;
+    l.precision = 38;
+    r.s128Value = 42000;
+    EXPECT_EQ(-1, datatypes::Decimal::compare(l, r));
+    // L values < R value, L scale > R scale
+    l.scale = 21;
+    l.precision = 38;
+    l.s128Value = 99;
+
+    r.scale = 20;
+    l.precision = 38;
+    r.s128Value = 420;
+    EXPECT_EQ(-1, datatypes::Decimal::compare(l, r));
+}
+
+TEST(Decimal, additionNoOverflowCheck)
+{
+    // Addition w/o overflow check
+    execplan::IDB_Decimal l, r, result;
+    // same precision, same scale, both positive values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(462, result.s128Value);
+    // same precision, same scale, both negative values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = -42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(-462, result.s128Value);
+    // same precision, same scale, +- values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(-378, result.s128Value);
+    // same precision, same scale, both 0
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 0;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 0;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(0, result.s128Value);
+    // diff scale
+    // same precision, L scale > R scale, both positive values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = 420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    int128_t s128ScaleMultiplier1 =
+        static_cast<int128_t>(10000000000000)*10000000000;
+    int128_t s128Result = r.s128Value*s128ScaleMultiplier1+l.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale > R scale, both negative values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = -42;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = r.s128Value*s128ScaleMultiplier1+l.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale > R scale, +- values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = r.s128Value*s128ScaleMultiplier1+l.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale > R scale, both 0
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 0;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = 0;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(0, result.s128Value);
+    // same precision, L scale < R scale, both positive values
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1+r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale < R scale, both negative values
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = -42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1+r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale < R scale, +- values
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1+r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale < R scale, both 0
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 0;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 0;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::addition<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1+r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+}
+
+TEST(Decimal, divisionNoOverflowCheck)
+{
+    // DIVISION
+    // same precision, same scale, both positive values
+    std::string decimalStr;
+    execplan::IDB_Decimal l, r, result;
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 43;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 420;
+    
+    result.scale = r.scale;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(r, l, result);
+    
+    EXPECT_EQ(r.scale, result.scale);
+    EXPECT_EQ(result.precision, result.precision);
+    EXPECT_EQ(r.s128Value/l.s128Value, result.s128Value);
+    // same precision, same scale, both negative values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = -42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = r.scale;
+    result.precision = r.precision;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(r, l, result);
+    EXPECT_EQ(r.scale, result.scale);
+    EXPECT_EQ(r.precision, result.precision);
+    EXPECT_EQ(r.s128Value/l.s128Value, result.s128Value);
+    // same precision, same scale, +- values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(l.s128Value/r.s128Value, result.s128Value);
+    // same precision, same scale, l = 0
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 0;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 42424242;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(0, result.s128Value);
+    // diff scale
+    // same precision, L scale > R scale, both positive values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = 420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(r, l, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    int128_t s128ScaleMultiplier1 =
+        static_cast<int128_t>(10000000000000)*10000000000;
+    int128_t s128Result = r.s128Value*s128ScaleMultiplier1/l.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale > R scale, both negative values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = -42;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(r, l, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = r.s128Value*s128ScaleMultiplier1/l.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale > R scale, +- values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(r, l, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = r.s128Value*s128ScaleMultiplier1/l.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale > R scale, L = 0
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 0;
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = 424242;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(0, result.s128Value);
+    // same precision, L scale > R scale, both MAX positive values
+    // WIP Investigate the next two
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 0; utils::int128Max(l.s128Value);
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = 0; utils::int128Max(r.s128Value);
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    // Use as an examplar
+    utils::int128Max(r.s128Value);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = r.s128Value*s128ScaleMultiplier1/l.s128Value;
+    // WIP
+    //EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale > R scale, both MIN negative values
+    l.scale = 38;
+    l.precision = 38;
+    l.s128Value = 0; utils::int128Min(l.s128Value);
+
+    r.scale = 15;
+    r.precision = 38;
+    r.s128Value = 0; utils::int128Min(l.s128Value);
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    //datatypes::Decimal::division<int128_t, false>(l, r, result);
+    // Use as an examplar
+    utils::int128Min(r.s128Value);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = r.s128Value*s128ScaleMultiplier1/l.s128Value;
+    //EXPECT_EQ(s128Result, result.s128Value);
+    // WIP
+    //EXPECT_EQ(r.s128Value, result.s128Value);
+
+    // same precision, L scale < R scale, both positive values
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1/r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale < R scale, both negative values
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = -42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1/r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale < R scale, +- values
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 42;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = -420;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1/r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale < R scale, L = 0
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 0;
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 42;
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1/r.s128Value;
+    EXPECT_EQ(s128Result, result.s128Value);
+    // same precision, L scale < R scale, both MAX positive values
+    // WIP Investigate the next two
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 0; utils::int128Max(l.s128Value);
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 0; utils::int128Max(r.s128Value);
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    datatypes::Decimal::division<int128_t, false>(l, r, result);
+    // Use as an examplar
+    utils::int128Max(r.s128Value);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1/r.s128Value;
+    // WIP
+    //EXPECT_EQ(s128Result, result.s128Value);
+    //EXPECT_EQ(r.s128Value, result.s128Value);
+    // same precision, L scale < R scale, both MIN negative values
+    l.scale = 15;
+    l.precision = 38;
+    l.s128Value = 0; utils::int128Min(l.s128Value);
+
+    r.scale = 38;
+    r.precision = 38;
+    r.s128Value = 0; utils::int128Min(l.s128Value);
+    
+    result.scale = 38;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    //datatypes::Decimal::division<int128_t, false>(l, r, result);
+    // Use as an examplar
+    utils::int128Min(r.s128Value);
+    EXPECT_EQ(38, result.scale);
+    EXPECT_EQ(38, result.precision);
+    s128Result = l.s128Value*s128ScaleMultiplier1/r.s128Value;
+    // WIP
+    // EXPECT_EQ(s128Result, result.s128Value);
+    //EXPECT_EQ(r.s128Value, result.s128Value);
+}
+
+void doDiv(execplan::IDB_Decimal& l,
+            execplan::IDB_Decimal& r,
+            execplan::IDB_Decimal& result)
+{
+    datatypes::Decimal::division<int128_t, true>(l, r, result);
+}
+
+TEST(Decimal, divisionWithOverflowCheck)
+{
+
+    // Divide max int128 by -1
+    execplan::IDB_Decimal l, r, result;
+    l.scale = 0;
+    l.precision = 38;
+    l.s128Value = datatypes::Decimal::maxInt128; 
+
+    r.scale = 0;
+    r.precision = 38;
+    r.s128Value = -1;
+    
+    result.scale = 0;
+    result.precision = 38;
+    result.s128Value = 42; 
+    EXPECT_THROW(doDiv(l, r, result), logging::OperationOverflowExcept);
+    // Divide two ints one of which overflows after the scaling.
+    l.scale = 0;
+    l.precision = 38;
+    l.s128Value = datatypes::Decimal::maxInt128; 
+
+    r.scale = 1;
+    r.precision = 38;
+    r.s128Value = 42;
+    
+    result.scale = 1;
+    result.precision = 38;
+    result.s128Value = 42; 
+    EXPECT_THROW(doDiv(l, r, result), logging::OperationOverflowExcept);
+    // Normal execution w/o overflow
+    l.scale = 0;
+    l.precision = 38;
+    l.s128Value = datatypes::Decimal::maxInt128-1;
+
+    r.scale = 0;
+    r.precision = 38;
+    r.s128Value = 0xFFFFFFFFFFFFFFFF;
+    
+    result.scale = 0;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    EXPECT_NO_THROW(doDiv(l, r, result));
+
+    l.s128Value /= r.s128Value;
+
+    EXPECT_EQ(0, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(l.s128Value, result.s128Value);
+}
+
+void doAdd(execplan::IDB_Decimal& l,
+            execplan::IDB_Decimal& r,
+            execplan::IDB_Decimal& result)
+{
+    datatypes::Decimal::addition<int128_t, true>(l, r, result);
+}
+
+TEST(Decimal, additionWithOverflowCheck)
+{
+    // Add two max ints
+    execplan::IDB_Decimal l, r, result;
+    l.scale = 0;
+    l.precision = 38;
+    l.s128Value = datatypes::Decimal::maxInt128-1;
+
+    r.scale = 0;
+    r.precision = 38;
+    r.s128Value = datatypes::Decimal::maxInt128-1;
+    
+    result.scale = 0;
+    result.precision = 38;
+    result.s128Value = 42; 
+    EXPECT_THROW(doAdd(l, r, result), logging::OperationOverflowExcept);
+    // Add two ints one of which overflows after the scaling.
+    l.scale = 0;
+    l.precision = 38;
+    l.s128Value = datatypes::Decimal::maxInt128-1;
+
+    r.scale = 1;
+    r.precision = 38;
+    r.s128Value = 0xFFFFFFFFFFFFFFFF;
+    
+    result.scale = 1;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    EXPECT_THROW(doAdd(l, r, result), logging::OperationOverflowExcept);
+    // Normal execution w/o overflow
+    l.scale = 0;
+    l.precision = 38;
+    l.s128Value = datatypes::Decimal::maxInt128-1;
+
+    r.scale = 0;
+    r.precision = 38;
+    r.s128Value = 0xFFFFFFFFFFFFFFFF;
+    
+    result.scale = 0;
+    result.precision = 38;
+    result.s128Value = 0;
+
+    EXPECT_NO_THROW(doDiv(l, r, result));
+
+    l.s128Value /= r.s128Value;
+
+    EXPECT_EQ(0, result.scale);
+    EXPECT_EQ(38, result.precision);
+    EXPECT_EQ(l.s128Value, result.s128Value);
+}

utils/common/columnwidth.h

@@ -18,7 +18,6 @@
 #ifndef UTILS_COLWIDTH_H
 #define UTILS_COLWIDTH_H
 
-#include "calpontsystemcatalog.h"
 #include "branchpred.h"
 
 namespace utils
@@ -36,13 +35,6 @@ namespace utils
         return width <= MAXLEGACYWIDTH;
     }
 
-    inline bool isWideDecimalType(const execplan::CalpontSystemCatalog::ColType& ct)
-    {
-        return ((ct.colDataType == execplan::CalpontSystemCatalog::DECIMAL ||
-            ct.colDataType == execplan::CalpontSystemCatalog::UDECIMAL) &&
-            ct.colWidth == MAXCOLUMNWIDTH);
-    }
-
     /** @brief Map a DECIMAL precision to data width in bytes */
     inline uint8_t widthByPrecision(unsigned p)
     {

utils/common/widedecimalutils.h

@@ -80,9 +80,7 @@ namespace utils
 
     inline void int128Min(int128_t& val)
     {
-        uint64_t* ptr = reinterpret_cast<uint64_t*>(&val);
+        val = int128_t(0x8000000000000000LL) << 64;
-        ptr[0] = 0;
-        ptr[1] = 0x8000000000000000;
     }
 
     inline void uint128Max(uint128_t& val)

utils/dataconvert/dataconvert.cpp

@@ -102,30 +102,6 @@ const string columnstore_big_precision[20] =
     "99999999999999999999999999999999999999"
 };
 
-const uint64_t columnstore_pow_10[20] =
-{
-    1ULL,
-    10ULL,
-    100ULL,
-    1000ULL,
-    10000ULL,
-    100000ULL,
-    1000000ULL,
-    10000000ULL,
-    100000000ULL,
-    1000000000ULL,
-    10000000000ULL,
-    100000000000ULL,
-    1000000000000ULL,
-    10000000000000ULL,
-    100000000000000ULL,
-    1000000000000000ULL,
-    10000000000000000ULL,
-    100000000000000000ULL,
-    1000000000000000000ULL,
-    10000000000000000000ULL
-};
-
 template <class T>
 bool from_string(T& t, const std::string& s, std::ios_base & (*f)(std::ios_base&))
 {
@@ -1257,33 +1233,30 @@ size_t DataConvert::writeIntPart(int128_t* dec,
 {
     int128_t intPart = *dec;
     int128_t high = 0, mid = 0, low = 0;
-    uint64_t div = 10000000000000000000ULL;
+    uint64_t maxUint64divisor = 10000000000000000000ULL;
 
     if (scale)
     {
-        const uint8_t maxPowOf10 =
-            (sizeof(columnstore_pow_10) / sizeof(columnstore_pow_10[0])) - 1;
-
         // Assuming scale = [0, 56]
-        switch (scale / maxPowOf10)
+        switch (scale / datatypes::maxPowOf10)
         {
             case 2: // scale = [38, 56]
-                intPart /= columnstore_pow_10[maxPowOf10];
+                intPart /= datatypes::mcs_pow_10[datatypes::maxPowOf10];
-                intPart /= columnstore_pow_10[maxPowOf10];
+                intPart /= datatypes::mcs_pow_10[datatypes::maxPowOf10];
                 low = intPart;
                 break;
             case 1: // scale = [19, 37]
-                intPart /= columnstore_pow_10[maxPowOf10];
+                intPart /= datatypes::mcs_pow_10[datatypes::maxPowOf10];
-                intPart /= columnstore_pow_10[scale % maxPowOf10];
+                intPart /= datatypes::mcs_pow_10[scale % datatypes::maxPowOf10];
-                low = intPart % div;
+                low = intPart % maxUint64divisor;
-                mid = intPart / div;
+                mid = intPart / maxUint64divisor;
                 break;
             case 0: // scale = [0, 18]
-                intPart /= columnstore_pow_10[scale % maxPowOf10];
+                intPart /= datatypes::mcs_pow_10[scale % datatypes::maxPowOf10];
-                low = intPart % div;
+                low = intPart % maxUint64divisor;
-                intPart /= div;
+                intPart /= maxUint64divisor;
-                mid = intPart % div;
+                mid = intPart % maxUint64divisor;
-                high = intPart / div;
+                high = intPart / maxUint64divisor;
                 break;
             default:
                 throw QueryDataExcept("writeIntPart() bad scale", formatErr);
@@ -1291,10 +1264,10 @@ size_t DataConvert::writeIntPart(int128_t* dec,
     }
     else
     {
-        low = intPart % div;
+        low = intPart % maxUint64divisor;
-        intPart /= div;
+        intPart /= maxUint64divisor;
-        mid = intPart % div;
+        mid = intPart % maxUint64divisor;
-        high = intPart / div;
+        high = intPart / maxUint64divisor;
     }
 
     // pod[0] is low 8 bytes, pod[1] is high 8 bytes
@@ -1337,19 +1310,16 @@ size_t DataConvert::writeFractionalPart(int128_t* dec,
 {
     int128_t scaleDivisor = 1;
 
-    const uint8_t maxPowOf10 =
+    switch (scale / datatypes::maxPowOf10)
-        (sizeof(columnstore_pow_10) / sizeof(columnstore_pow_10[0])) - 1;
-
-    switch (scale / maxPowOf10)
     {
         case 2:
-            scaleDivisor *= columnstore_pow_10[maxPowOf10];
+            scaleDivisor *= datatypes::mcs_pow_10[datatypes::maxPowOf10];
-            scaleDivisor *= columnstore_pow_10[maxPowOf10];
+            scaleDivisor *= datatypes::mcs_pow_10[datatypes::maxPowOf10];
             break;
         case 1:
-            scaleDivisor *= columnstore_pow_10[maxPowOf10];
+            scaleDivisor *= datatypes::mcs_pow_10[datatypes::maxPowOf10];
         case 0:
-            scaleDivisor *= columnstore_pow_10[scale % maxPowOf10];
+            scaleDivisor *= datatypes::mcs_pow_10[scale % datatypes::maxPowOf10];
     }
 
     int128_t fractionalPart = *dec % scaleDivisor;

utils/funcexp/funcexp.cpp

@@ -42,6 +42,8 @@ using namespace joblist;
 #include "../udfsdk/udfsdk.h"
 #endif
 
+#include "mcs_decimal.h"
+
 namespace funcexp
 {
 
@@ -471,13 +473,22 @@ void FuncExp::evaluate(rowgroup::Row& row, std::vector<execplan::SRCP>& expressi
             case CalpontSystemCatalog::UDECIMAL:
             {
                 IDB_Decimal val = expression[i]->getDecimalVal(row, isNull);
-
+                if (expression[i]->resultType().colWidth
-                // WIP check for null and overflow here.
+                    == datatypes::MAXDECIMALWIDTH)
-                if (expression[i]->resultType().colWidth == 16)
                 {
-                    row.setBinaryField_offset(&val.s128Value,
+                    if (isNull)
-                        expression[i]->resultType().colWidth,
+                    {
-                        row.getOffset(expression[i]->outputIndex()));
+                         row.setBinaryField_offset(
+                            const_cast<int128_t*>(&datatypes::Decimal128Null),
+                            expression[i]->resultType().colWidth,
+                            row.getOffset(expression[i]->outputIndex()));
+                    }
+                    else
+                    {
+                        row.setBinaryField_offset(&val.s128Value,
+                            expression[i]->resultType().colWidth,
+                            row.getOffset(expression[i]->outputIndex()));
+                    }
                 }
                 else
                 {

utils/loggingcpp/exceptclasses.h

@@ -187,6 +187,16 @@ public:
         std::runtime_error(msg) { }
 };
 
+/** @brief Exception for F&E framework to throw on op overflow
+ *  Invalid Operation Exception
+ */
+class OperationOverflowExcept : public std::runtime_error
+{
+public:
+    /** Takes a character string describing the error.  */
+    OperationOverflowExcept(const std::string& msg) :
+        std::runtime_error(msg) { }
+};
 /** @brief specific error exception class for getSysData in Calpontsystemcatalog.
 *   @bug 2574
 *

utils/rowgroup/rowgroup.cpp

@@ -1629,7 +1629,6 @@ void applyMapping(const int* mapping, const Row& in, Row* out)
                 out->setVarBinaryField(in.getVarBinaryField(i), in.getVarBinaryLength(i), mapping[i]);
             else if (UNLIKELY(in.isLongString(i)))
                 out->setStringField(in.getStringPointer(i), in.getStringLength(i), mapping[i]);
-            //out->setStringField(in.getStringField(i), mapping[i]);
             else if (UNLIKELY(in.isShortString(i)))
                 out->setUintField(in.getUintField(i), mapping[i]);
             else if (UNLIKELY(in.getColTypes()[i] == execplan::CalpontSystemCatalog::LONGDOUBLE))