Fix decimal floor/ceil (#10365)

Author: ChangRui-Ryan

dbms/src/Functions/FunctionsRound.h

@@ -199,77 +199,6 @@ enum class RoundingMode
 #endif
 };
 
-/** Rounding functions for decimal values
- */
-
-template <typename T, RoundingMode rounding_mode, ScaleMode scale_mode, typename OutputType>
-struct DecimalRoundingComputation
-{
-    static_assert(IsDecimal<T>);
-    static const size_t data_count = 1;
-    static size_t prepare(size_t scale) { return scale; }
-    // compute need decimal_scale to interpret decimals
-    static inline void compute(
-        const T * __restrict in,
-        size_t scale,
-        OutputType * __restrict out,
-        ScaleType decimal_scale)
-    {
-        static_assert(std::is_same_v<T, OutputType> || std::is_same_v<OutputType, Int64>);
-        Float64 val = in->template toFloat<Float64>(decimal_scale);
-
-        if constexpr (scale_mode == ScaleMode::Positive)
-        {
-            val = val * scale;
-        }
-        else if constexpr (scale_mode == ScaleMode::Negative)
-        {
-            val = val / scale;
-        }
-
-        if constexpr (rounding_mode == RoundingMode::Round)
-        {
-            val = round(val);
-        }
-        else if constexpr (rounding_mode == RoundingMode::Floor)
-        {
-            val = floor(val);
-        }
-        else if constexpr (rounding_mode == RoundingMode::Ceil)
-        {
-            val = ceil(val);
-        }
-        else if constexpr (rounding_mode == RoundingMode::Trunc)
-        {
-            val = trunc(val);
-        }
-
-
-        if constexpr (scale_mode == ScaleMode::Positive)
-        {
-            val = val / scale;
-        }
-        else if constexpr (scale_mode == ScaleMode::Negative)
-        {
-            val = val * scale;
-        }
-
-        if constexpr (std::is_same_v<T, OutputType>)
-        {
-            *out = ToDecimal<Float64, T>(val, decimal_scale);
-        }
-        else if constexpr (std::is_same_v<OutputType, Int64>)
-        {
-            *out = static_cast<Int64>(val);
-        }
-        else
-        {
-            ; // never arrived here
-        }
-    }
-};
-
-
 /** Rounding functions for integer values.
   */
 template <typename T, RoundingMode rounding_mode, ScaleMode scale_mode>
@@ -327,12 +256,75 @@ struct IntegerRoundingComputation
         }
     }
 
-    static ALWAYS_INLINE void compute(const T * __restrict in, size_t scale, T * __restrict out)
+    static ALWAYS_INLINE void compute(const T * __restrict in, T scale, T * __restrict out)
     {
         *out = compute(*in, scale);
     }
 };
 
+/** Rounding functions for decimal values
+ */
+
+template <typename T, RoundingMode rounding_mode, ScaleMode scale_mode, typename OutputType>
+struct DecimalRoundingComputation
+{
+    static_assert(IsDecimal<T>);
+    static const size_t data_count = 1;
+    static size_t prepare(size_t scale) { return scale; }
+    // compute need decimal_scale to interpret decimals
+    static inline void compute(
+        const T * __restrict in,
+        size_t scale,
+        OutputType * __restrict out,
+        ScaleType decimal_scale)
+    {
+        static_assert(std::is_same_v<T, OutputType> || std::is_same_v<OutputType, Int64>);
+        using NativeType = T::NativeType;
+        // Currently, we only use DecimalRoundingComputation for floor/ceil.
+        // As for round/truncate, we always use tidbRoundWithFrac/tidbTruncateWithFrac.
+        // So, we only handle ScaleMode::Zero here.
+        if constexpr (scale_mode == ScaleMode::Zero)
+        {
+            auto scale_factor = intExp10OfSize<NativeType>(decimal_scale);
+            try
+            {
+                if constexpr (rounding_mode == RoundingMode::Floor)
+                {
+                    auto x = in->value;
+                    if (x < 0)
+                        x -= scale_factor - 1;
+                    *out = static_cast<OutputType>(x / scale_factor);
+                }
+                else if constexpr (rounding_mode == RoundingMode::Ceil)
+                {
+                    auto x = in->value;
+                    if (x >= 0)
+                        x += scale_factor - 1;
+                    *out = static_cast<OutputType>(x / scale_factor);
+                }
+                else
+                {
+                    throw Exception(
+                        "Logical error: unexpected 'rounding_mode' of DecimalRoundingComputation",
+                        ErrorCodes::LOGICAL_ERROR);
+                }
+            }
+            catch (const std::overflow_error & e)
+            {
+                throw Exception(
+                    "Logical error: unexpected overflow in DecimalRoundingComputation",
+                    ErrorCodes::LOGICAL_ERROR);
+            }
+        }
+        else
+        {
+            throw Exception(
+                "Logical error: unexpected 'scale_mode' of DecimalRoundingComputation and unexpected scale: "
+                    + toString(scale),
+                ErrorCodes::LOGICAL_ERROR);
+        }
+    }
+};
 
 #if __SSE4_1__
 
@@ -540,7 +532,7 @@ struct IntegerRoundingImpl
 
         while (p_in < end_in)
         {
-            Op::compute(p_in, scale, p_out);
+            Op::compute(p_in, static_cast<T>(scale), p_out);
             ++p_in;
             ++p_out;
         }
@@ -698,7 +690,12 @@ struct Dispatcher
 
         if constexpr (IsDecimal<OutputType>)
         {
-            auto col_res = ColumnDecimal<OutputType>::create(col->getData().size(), col->getData().getScale());
+            UInt32 res_scale = 0;
+            if constexpr (rounding_mode == RoundingMode::Round || rounding_mode == RoundingMode::Trunc)
+            {
+                res_scale = col->getData().getScale();
+            }
+            auto col_res = ColumnDecimal<OutputType>::create(col->getData().size(), res_scale);
             typename ColumnDecimal<OutputType>::Container & vec_res = col_res->getData();
             applyInternal(col, vec_res, col_res, block, scale_arg, result);
         }
@@ -813,6 +810,20 @@ class FunctionRounding : public IFunction
                     fmt::format("Illegal type {} of argument of function {}", arguments[0]->getName(), getName()),
                     ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
 
+        if constexpr (rounding_mode == RoundingMode::Ceil || rounding_mode == RoundingMode::Floor)
+        {
+            if (arguments[0]->isDecimal())
+            {
+                if (const auto * decimal_type32 = checkAndGetDataType<DataTypeDecimal32>(arguments[0].get()))
+                    return std::make_shared<DataTypeDecimal32>(decimal_type32->getPrec(), 0);
+                else if (const auto * decimal_type64 = checkAndGetDataType<DataTypeDecimal64>(arguments[0].get()))
+                    return std::make_shared<DataTypeDecimal64>(decimal_type64->getPrec(), 0);
+                else if (const auto * decimal_type128 = checkAndGetDataType<DataTypeDecimal128>(arguments[0].get()))
+                    return std::make_shared<DataTypeDecimal128>(decimal_type128->getPrec(), 0);
+                else if (const auto * decimal_type256 = checkAndGetDataType<DataTypeDecimal256>(arguments[0].get()))
+                    return std::make_shared<DataTypeDecimal256>(decimal_type256->getPrec(), 0);
+            }
+        }
         return arguments[0];
     }