From 34ead026d4d4639aefcfb0083e8e920f4cc4b47d Mon Sep 17 00:00:00 2001
From: Gabriel <gabrielleebuaa@gmail.com>
Date: Wed, 22 Mar 2023 11:35:07 +0800
Subject: [PATCH] [Improvement](decimal) Improve cast function between decimal
 type (#17996)

---
 be/src/vec/data_types/data_type_decimal.h | 60 ++++++++++++++++
 be/src/vec/functions/function_cast.h      | 83 +++++++++++++----------
 2 files changed, 106 insertions(+), 37 deletions(-)

diff --git a/be/src/vec/data_types/data_type_decimal.h b/be/src/vec/data_types/data_type_decimal.h
index c330d0f386..9123b3837f 100644
--- a/be/src/vec/data_types/data_type_decimal.h
+++ b/be/src/vec/data_types/data_type_decimal.h
@@ -346,6 +346,66 @@ convert_decimals(const typename FromDataType::FieldType& value, UInt32 scale_fro
     return converted_value;
 }
 
+template <typename FromDataType, typename ToDataType>
+void convert_decimal_cols(
+        const typename ColumnDecimal<typename FromDataType::FieldType>::Container& vec_from,
+        typename ColumnDecimal<typename ToDataType::FieldType>::Container& vec_to,
+        UInt32 scale_from, UInt32 scale_to, UInt8* overflow_flag = nullptr) {
+    using FromFieldType = typename FromDataType::FieldType;
+    using ToFieldType = typename ToDataType::FieldType;
+    using MaxFieldType =
+            std::conditional_t<(sizeof(FromFieldType) == sizeof(ToFieldType)) &&
+                                       (std::is_same_v<ToFieldType, Decimal128I> ||
+                                        std::is_same_v<FromFieldType, Decimal128I>),
+                               Decimal128I,
+                               std::conditional_t<(sizeof(FromFieldType) > sizeof(ToFieldType)),
+                                                  FromFieldType, ToFieldType>>;
+    using MaxNativeType = typename MaxFieldType::NativeType;
+
+    if (scale_to > scale_from) {
+        MaxNativeType multiplier =
+                DataTypeDecimal<MaxFieldType>::get_scale_multiplier(scale_to - scale_from);
+        MaxNativeType res;
+        for (size_t i = 0; i < vec_from.size(); i++) {
+            if (common::mul_overflow(static_cast<MaxNativeType>(vec_from[i]), multiplier, res)) {
+                if (overflow_flag) {
+                    overflow_flag[i] = 1;
+                }
+                VLOG_DEBUG << "Decimal convert overflow";
+                vec_to[i] = res < 0 ? std::numeric_limits<typename ToFieldType::NativeType>::min()
+                                    : std::numeric_limits<typename ToFieldType::NativeType>::max();
+            } else {
+                vec_to[i] = res;
+            }
+        }
+    } else {
+        MaxNativeType multiplier =
+                DataTypeDecimal<MaxFieldType>::get_scale_multiplier(scale_from - scale_to);
+        for (size_t i = 0; i < vec_from.size(); i++) {
+            vec_to[i] = vec_from[i] / multiplier;
+        }
+    }
+
+    if constexpr (sizeof(FromFieldType) > sizeof(ToFieldType)) {
+        for (size_t i = 0; i < vec_from.size(); i++) {
+            if (vec_to[i] < std::numeric_limits<typename ToFieldType::NativeType>::min()) {
+                if (overflow_flag) {
+                    *overflow_flag = 1;
+                }
+                VLOG_DEBUG << "Decimal convert overflow";
+                vec_to[i] = std::numeric_limits<typename ToFieldType::NativeType>::min();
+            }
+            if (vec_to[i] > std::numeric_limits<typename ToFieldType::NativeType>::max()) {
+                if (overflow_flag) {
+                    *overflow_flag = 1;
+                }
+                VLOG_DEBUG << "Decimal convert overflow";
+                vec_to[i] = std::numeric_limits<typename ToFieldType::NativeType>::max();
+            }
+        }
+    }
+}
+
 template <typename FromDataType, typename ToDataType>
 std::enable_if_t<IsDataTypeDecimal<FromDataType> && IsDataTypeNumber<ToDataType>,
                  typename ToDataType::FieldType>
diff --git a/be/src/vec/functions/function_cast.h b/be/src/vec/functions/function_cast.h
index 633a76456e..d754d80e20 100644
--- a/be/src/vec/functions/function_cast.h
+++ b/be/src/vec/functions/function_cast.h
@@ -113,45 +113,54 @@ struct ConvertImpl {
                     col_null_map_to = ColumnUInt8::create(size, 0);
                     vec_null_map_to = col_null_map_to->get_data().data();
                 }
-                for (size_t i = 0; i < size; ++i) {
-                    if constexpr (IsDataTypeDecimal<FromDataType> &&
-                                  IsDataTypeDecimal<ToDataType>) {
-                        vec_to[i] = convert_decimals<FromDataType, ToDataType>(
-                                vec_from[i], vec_from.get_scale(), vec_to.get_scale(),
-                                vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
-                    } else if constexpr (IsDataTypeDecimal<FromDataType> &&
-                                         IsDataTypeNumber<ToDataType>) {
-                        vec_to[i] = convert_from_decimal<FromDataType, ToDataType>(
-                                vec_from[i], vec_from.get_scale());
-                    } else if constexpr (IsDataTypeNumber<FromDataType> &&
-                                         IsDataTypeDecimal<ToDataType>) {
-                        vec_to[i] = convert_to_decimal<FromDataType, ToDataType>(
-                                vec_from[i], vec_to.get_scale(),
-                                vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
-                    } else if constexpr (IsTimeType<FromDataType> &&
-                                         IsDataTypeDecimal<ToDataType>) {
-                        vec_to[i] = convert_to_decimal<DataTypeInt64, ToDataType>(
-                                reinterpret_cast<const VecDateTimeValue&>(vec_from[i]).to_int64(),
-                                vec_to.get_scale(),
-                                vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
-                    } else if constexpr (IsDateV2Type<FromDataType> &&
-                                         IsDataTypeDecimal<ToDataType>) {
-                        vec_to[i] = convert_to_decimal<DataTypeUInt32, ToDataType>(
-                                reinterpret_cast<const DateV2Value<DateV2ValueType>&>(vec_from[i])
-                                        .to_date_int_val(),
-                                vec_to.get_scale(),
-                                vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
-                    } else if constexpr (IsDateTimeV2Type<FromDataType> &&
-                                         IsDataTypeDecimal<ToDataType>) {
-                        // TODO: should we consider the scale of datetimev2?
-                        vec_to[i] = convert_to_decimal<DataTypeUInt64, ToDataType>(
-                                reinterpret_cast<const DateV2Value<DateTimeV2ValueType>&>(
-                                        vec_from[i])
-                                        .to_date_int_val(),
-                                vec_to.get_scale(),
-                                vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
+                if constexpr (IsDataTypeDecimal<FromDataType> && IsDataTypeDecimal<ToDataType>) {
+                    convert_decimal_cols<FromDataType, ToDataType>(
+                            vec_from, vec_to, vec_from.get_scale(), vec_to.get_scale(),
+                            vec_null_map_to);
+                } else {
+                    for (size_t i = 0; i < size; ++i) {
+                        if constexpr (IsDataTypeDecimal<FromDataType> &&
+                                      IsDataTypeDecimal<ToDataType>) {
+                            vec_to[i] = convert_decimals<FromDataType, ToDataType>(
+                                    vec_from[i], vec_from.get_scale(), vec_to.get_scale(),
+                                    vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
+                        } else if constexpr (IsDataTypeDecimal<FromDataType> &&
+                                             IsDataTypeNumber<ToDataType>) {
+                            vec_to[i] = convert_from_decimal<FromDataType, ToDataType>(
+                                    vec_from[i], vec_from.get_scale());
+                        } else if constexpr (IsDataTypeNumber<FromDataType> &&
+                                             IsDataTypeDecimal<ToDataType>) {
+                            vec_to[i] = convert_to_decimal<FromDataType, ToDataType>(
+                                    vec_from[i], vec_to.get_scale(),
+                                    vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
+                        } else if constexpr (IsTimeType<FromDataType> &&
+                                             IsDataTypeDecimal<ToDataType>) {
+                            vec_to[i] = convert_to_decimal<DataTypeInt64, ToDataType>(
+                                    reinterpret_cast<const VecDateTimeValue&>(vec_from[i])
+                                            .to_int64(),
+                                    vec_to.get_scale(),
+                                    vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
+                        } else if constexpr (IsDateV2Type<FromDataType> &&
+                                             IsDataTypeDecimal<ToDataType>) {
+                            vec_to[i] = convert_to_decimal<DataTypeUInt32, ToDataType>(
+                                    reinterpret_cast<const DateV2Value<DateV2ValueType>&>(
+                                            vec_from[i])
+                                            .to_date_int_val(),
+                                    vec_to.get_scale(),
+                                    vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
+                        } else if constexpr (IsDateTimeV2Type<FromDataType> &&
+                                             IsDataTypeDecimal<ToDataType>) {
+                            // TODO: should we consider the scale of datetimev2?
+                            vec_to[i] = convert_to_decimal<DataTypeUInt64, ToDataType>(
+                                    reinterpret_cast<const DateV2Value<DateTimeV2ValueType>&>(
+                                            vec_from[i])
+                                            .to_date_int_val(),
+                                    vec_to.get_scale(),
+                                    vec_null_map_to ? &vec_null_map_to[i] : vec_null_map_to);
+                        }
                     }
                 }
+
                 if (check_overflow) {
                     block.replace_by_position(
                             result,