[fix](round) fix round decimal128 overflow (apache#37733) (apache#37965)

be/src/vec/functions/round.h

@@ -22,13 +22,15 @@
 
 #include "vec/columns/column_const.h"
 #include "vec/columns/columns_number.h"
+#include "vec/exec/format/format_common.h"
 #include "vec/functions/function.h"
 #if defined(__SSE4_1__) || defined(__aarch64__)
 #include "util/sse_util.hpp"
 #else
 #include <fenv.h>
 #endif
 #include <algorithm>
+#include <type_traits>
 
 #include "vec/columns/column.h"
 #include "vec/columns/column_decimal.h"
@@ -64,7 +66,7 @@ enum class TieBreakingMode {
 };
 
 template <typename T, RoundingMode rounding_mode, ScaleMode scale_mode,
-          TieBreakingMode tie_breaking_mode>
+          TieBreakingMode tie_breaking_mode, typename U>
 struct IntegerRoundingComputation {
     static const size_t data_count = 1;
 
@@ -116,7 +118,7 @@ struct IntegerRoundingComputation {
         __builtin_unreachable();
     }
 
-    static ALWAYS_INLINE T compute(T x, T scale, size_t target_scale) {
+    static ALWAYS_INLINE T compute(T x, T scale, T target_scale) {
         switch (scale_mode) {
         case ScaleMode::Zero:
         case ScaleMode::Positive:
@@ -128,10 +130,10 @@ struct IntegerRoundingComputation {
         __builtin_unreachable();
     }
 
-    static ALWAYS_INLINE void compute(const T* __restrict in, size_t scale, T* __restrict out,
-                                      size_t target_scale) {
+    static ALWAYS_INLINE void compute(const T* __restrict in, U scale, T* __restrict out,
+                                      U target_scale) {
         if constexpr (sizeof(T) <= sizeof(scale) && scale_mode == ScaleMode::Negative) {
-            if (scale > size_t(std::numeric_limits<T>::max())) {
+            if (scale >= std::numeric_limits<T>::max()) {
                 *out = 0;
                 return;
             }
@@ -145,23 +147,24 @@ class DecimalRoundingImpl {
 private:
     using NativeType = typename T::NativeType;
     using Op = IntegerRoundingComputation<NativeType, rounding_mode, ScaleMode::Negative,
-                                          tie_breaking_mode>;
+                                          tie_breaking_mode, NativeType>;
     using Container = typename ColumnDecimal<T>::Container;
 
 public:
     static NO_INLINE void apply(const Container& in, UInt32 in_scale, Container& out,
                                 Int16 out_scale) {
         Int16 scale_arg = in_scale - out_scale;
         if (scale_arg > 0) {
-            size_t scale = int_exp10(scale_arg);
+            auto scale = DecimalScaleParams::get_scale_factor<NativeType>(scale_arg);
 
             const NativeType* __restrict p_in = reinterpret_cast<const NativeType*>(in.data());
             const NativeType* end_in = reinterpret_cast<const NativeType*>(in.data()) + in.size();
             NativeType* __restrict p_out = reinterpret_cast<NativeType*>(out.data());
 
             if (out_scale < 0) {
+                auto negative_scale = DecimalScaleParams::get_scale_factor<NativeType>(-out_scale);
                 while (p_in < end_in) {
-                    Op::compute(p_in, scale, p_out, int_exp10(-out_scale));
+                    Op::compute(p_in, scale, p_out, negative_scale);
                     ++p_in;
                     ++p_out;
                 }
@@ -320,7 +323,7 @@ template <typename T, RoundingMode rounding_mode, ScaleMode scale_mode,
           TieBreakingMode tie_breaking_mode>
 struct IntegerRoundingImpl {
 private:
-    using Op = IntegerRoundingComputation<T, rounding_mode, scale_mode, tie_breaking_mode>;
+    using Op = IntegerRoundingComputation<T, rounding_mode, scale_mode, tie_breaking_mode, size_t>;
     using Container = typename ColumnVector<T>::Container;
 
 public:

regression-test/data/nereids_p0/sql_functions/math_functions/test_round_overflow.out

@@ -0,0 +1,10 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select1 --
+186
+
+-- !select2 --
+0
+
+-- !select3 --
+20000000000000000000000
+

regression-test/suites/nereids_p0/sql_functions/math_functions/test_round_overflow.groovy

@@ -0,0 +1,25 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_round_overflow") {
+    sql "SET enable_nereids_planner=true"
+    sql "SET enable_fallback_to_original_planner=false"
+
+    qt_select1 "select round(coalesce(186,-33280029.8473323000000000000000000));"
+    qt_select2 "select round(coalesce(186,-33280029.8473323000000000000000000), -20);"
+    qt_select3 "select round(coalesce(18618500001234567890123.0,-33280029.847332300000000),-22);"
+}