diff --git a/be/src/exprs/function/cast/cast_base.h b/be/src/exprs/function/cast/cast_base.h
index cc58e29d4acd11..a34acbcf527a59 100644
--- a/be/src/exprs/function/cast/cast_base.h
+++ b/be/src/exprs/function/cast/cast_base.h
@@ -157,7 +157,7 @@ WrapperType create_identity_wrapper(const DataTypePtr&);
 
 } // namespace CastWrapper
 
-enum class CastModeType { StrictMode, NonStrictMode };
+enum class CastModeType { StrictMode, NonStrictMode, LosslessMode, StrictLosslessMode };
 
 inline std::string cast_mode_type_to_string(CastModeType cast_mode) {
     switch (cast_mode) {
@@ -165,6 +165,10 @@ inline std::string cast_mode_type_to_string(CastModeType cast_mode) {
         return "StrictMode";
     case CastModeType::NonStrictMode:
         return "NonStrictMode";
+    case CastModeType::LosslessMode:
+        return "LosslessMode";
+    case CastModeType::StrictLosslessMode:
+        return "StrictLosslessMode";
     default:
         return "Unknown";
     }
diff --git a/be/src/exprs/function/cast/cast_to_decimal.h b/be/src/exprs/function/cast/cast_to_decimal.h
index 8dc4c6b4aeac93..a35c1bbc47d92d 100644
--- a/be/src/exprs/function/cast/cast_to_decimal.h
+++ b/be/src/exprs/function/cast/cast_to_decimal.h
@@ -17,7 +17,11 @@
 
 #pragma once
 
+#include <algorithm>
+#include <cctype>
 #include <cmath>
+#include <cstdint>
+#include <limits>
 #include <type_traits>
 
 #include "common/status.h"
@@ -35,45 +39,174 @@ namespace doris {
                    value, from_type_name, precision, scale))
 
 struct CastToDecimal {
+    static bool is_lossless_decimal_string(const char* data, size_t size, UInt32 precision,
+                                           UInt32 target_scale) {
+        size_t begin = 0;
+        size_t end = size;
+        while (begin < end && std::isspace(static_cast<unsigned char>(data[begin]))) {
+            ++begin;
+        }
+        while (begin < end && std::isspace(static_cast<unsigned char>(data[end - 1]))) {
+            --end;
+        }
+        if (begin == end) {
+            return false;
+        }
+
+        size_t pos = begin;
+        if (data[pos] == '+' || data[pos] == '-') {
+            ++pos;
+        }
+
+        bool seen_dot = false;
+        bool seen_digit = false;
+        size_t digits_before_dot = 0;
+        size_t digits = 0;
+        size_t first_non_zero = 0;
+        size_t last_non_zero = 0;
+        while (pos < end && data[pos] != 'e' && data[pos] != 'E') {
+            unsigned char ch = static_cast<unsigned char>(data[pos]);
+            if (std::isdigit(ch)) {
+                seen_digit = true;
+                ++digits;
+                if (!seen_dot) {
+                    ++digits_before_dot;
+                }
+                if (ch != '0') {
+                    if (first_non_zero == 0) {
+                        first_non_zero = digits;
+                    }
+                    last_non_zero = digits;
+                }
+            } else if (data[pos] == '.' && !seen_dot) {
+                seen_dot = true;
+            } else {
+                return false;
+            }
+            ++pos;
+        }
+        if (!seen_digit) {
+            return false;
+        }
+
+        int64_t exponent = 0;
+        if (pos < end) {
+            ++pos;
+            bool negative_exponent = false;
+            if (pos < end && (data[pos] == '+' || data[pos] == '-')) {
+                negative_exponent = data[pos] == '-';
+                ++pos;
+            }
+            if (pos == end) {
+                return false;
+            }
+
+            int64_t exponent_magnitude = 0;
+            bool exponent_overflow = false;
+            while (pos < end) {
+                unsigned char ch = static_cast<unsigned char>(data[pos]);
+                if (!std::isdigit(ch)) {
+                    return false;
+                }
+                if (!exponent_overflow) {
+                    if (exponent_magnitude >
+                        (std::numeric_limits<int64_t>::max() - (ch - '0')) / 10) {
+                        exponent_overflow = true;
+                    } else {
+                        exponent_magnitude = exponent_magnitude * 10 + (ch - '0');
+                    }
+                }
+                ++pos;
+            }
+            if (exponent_overflow) {
+                return first_non_zero == 0;
+            }
+            exponent = negative_exponent ? -exponent_magnitude : exponent_magnitude;
+        }
+
+        // Zero is exactly representable at every decimal precision and scale.
+        if (first_non_zero == 0) {
+            return true;
+        }
+
+        size_t first_non_zero_offset = first_non_zero - 1;
+        if (digits_before_dot > std::numeric_limits<int64_t>::max() ||
+            first_non_zero_offset > std::numeric_limits<int64_t>::max()) {
+            return false;
+        }
+        int64_t base_decimal_point = static_cast<int64_t>(digits_before_dot) -
+                                     static_cast<int64_t>(first_non_zero_offset);
+
+        size_t significant_digits = last_non_zero - first_non_zero + 1;
+        if (significant_digits > precision) {
+            return false;
+        }
+
+        if ((exponent > 0 && base_decimal_point > std::numeric_limits<int64_t>::max() - exponent) ||
+            (exponent < 0 && base_decimal_point < std::numeric_limits<int64_t>::min() - exponent)) {
+            return false;
+        }
+        int64_t decimal_point = base_decimal_point + exponent;
+        int64_t max_integer_digits = precision - target_scale;
+        if (decimal_point > max_integer_digits ||
+            decimal_point < -static_cast<int64_t>(target_scale)) {
+            return false;
+        }
+        int64_t required_scale =
+                std::max(static_cast<int64_t>(significant_digits) - decimal_point, int64_t {0});
+        return required_scale <= target_scale;
+    }
+
     template <typename ToCppT>
         requires(IsDecimalNumber<ToCppT>)
-    static inline bool from_string(const StringRef& from, ToCppT& to, UInt32 precision,
-                                   UInt32 scale, CastParameters& params) {
+    static inline StringParser::ParseResult parse_string(const StringRef& from, ToCppT& to,
+                                                         UInt32 precision, UInt32 scale) {
+        StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
         if constexpr (IsDecimalV2<ToCppT>) {
-            StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
             to = DecimalV2Value(StringParser::string_to_decimal<TYPE_DECIMALV2>(
-                    from.data, (int)from.size, DecimalV2Value::PRECISION, DecimalV2Value::SCALE,
+                    from.data, from.size, DecimalV2Value::PRECISION, DecimalV2Value::SCALE,
                     &result));
-            return result == StringParser::PARSE_SUCCESS;
         }
 
         if constexpr (IsDecimal32<ToCppT>) {
-            StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
-            to.value = StringParser::string_to_decimal<TYPE_DECIMAL32>(from.data, (int)from.size,
+            to.value = StringParser::string_to_decimal<TYPE_DECIMAL32>(from.data, from.size,
                                                                        precision, scale, &result);
-            return result == StringParser::PARSE_SUCCESS;
         }
 
         if constexpr (IsDecimal64<ToCppT>) {
-            StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
-            to.value = StringParser::string_to_decimal<TYPE_DECIMAL64>(from.data, (int)from.size,
+            to.value = StringParser::string_to_decimal<TYPE_DECIMAL64>(from.data, from.size,
                                                                        precision, scale, &result);
-            return result == StringParser::PARSE_SUCCESS;
         }
 
         if constexpr (IsDecimal128V3<ToCppT>) {
-            StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
-            to.value = StringParser::string_to_decimal<TYPE_DECIMAL128I>(from.data, (int)from.size,
+            to.value = StringParser::string_to_decimal<TYPE_DECIMAL128I>(from.data, from.size,
                                                                          precision, scale, &result);
-            return result == StringParser::PARSE_SUCCESS;
         }
 
         if constexpr (IsDecimal256<ToCppT>) {
-            StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
-            to.value = StringParser::string_to_decimal<TYPE_DECIMAL256>(from.data, (int)from.size,
+            to.value = StringParser::string_to_decimal<TYPE_DECIMAL256>(from.data, from.size,
                                                                         precision, scale, &result);
-            return result == StringParser::PARSE_SUCCESS;
         }
+        return result;
+    }
+
+    template <typename ToCppT>
+        requires(IsDecimalNumber<ToCppT>)
+    static inline bool from_string(const StringRef& from, ToCppT& to, UInt32 precision,
+                                   UInt32 scale, CastParameters& params) {
+        return parse_string(from, to, precision, scale) == StringParser::PARSE_SUCCESS;
+    }
+
+    template <typename ToCppT>
+        requires(IsDecimalNumber<ToCppT>)
+    static inline bool from_string_lossless(const StringRef& from, ToCppT& to, UInt32 precision,
+                                            UInt32 scale, CastParameters& params) {
+        if (!is_lossless_decimal_string(from.data, from.size, precision, scale)) {
+            return false;
+        }
+        StringParser::ParseResult result = parse_string(from, to, precision, scale);
+        return result == StringParser::PARSE_SUCCESS || result == StringParser::PARSE_OVERFLOW ||
+               result == StringParser::PARSE_UNDERFLOW;
     }
 
     // cast int to decimal
@@ -704,6 +837,49 @@ class CastToImpl<Mode, DataTypeString, ToDataType> : public CastToBase {
             RETURN_IF_ERROR(
                     serde->from_string_strict_mode_batch(*col_from, *column_to, {}, null_map));
             block.get_by_position(result).column = std::move(column_to);
+        } else if constexpr (Mode == CastModeType::LosslessMode ||
+                             Mode == CastModeType::StrictLosslessMode) {
+            auto nullable_col_to = create_empty_nullable_column(nested_to_type);
+            auto& nullable_to = *nullable_col_to;
+            nullable_to.resize(col_from->size());
+            auto& column_to =
+                    assert_cast<typename ToDataType::ColumnType&>(nullable_to.get_nested_column());
+            auto& values_to = column_to.get_data();
+            auto& null_map_to = nullable_to.get_null_map_data();
+            auto* decimal_to_type = assert_cast<const ToDataType*>(nested_to_type.get());
+            const auto& chars = col_from->get_chars();
+            const auto& offsets = col_from->get_offsets();
+            size_t current_offset = 0;
+            UInt32 precision = decimal_to_type->get_precision();
+            UInt32 scale = decimal_to_type->get_scale();
+            [[maybe_unused]] CastParameters params;
+            params.is_strict = false;
+            for (size_t i = 0; i < col_from->size(); ++i) {
+                size_t next_offset = offsets[i];
+                size_t string_size = next_offset - current_offset;
+                bool source_is_null = null_map != nullptr && null_map[i];
+                values_to[i] = {};
+                StringRef source(reinterpret_cast<const char*>(&chars[current_offset]),
+                                 string_size);
+                bool parsed = false;
+                if constexpr (Mode == CastModeType::StrictLosslessMode) {
+                    if (!source_is_null &&
+                        CastToDecimal::parse_string(source, values_to[i], precision, scale) !=
+                                StringParser::PARSE_SUCCESS) {
+                        return Status::InvalidArgument("parse number fail, string: '{}'",
+                                                       std::string(source.data, source.size));
+                    }
+                    parsed = !source_is_null && CastToDecimal::is_lossless_decimal_string(
+                                                        source.data, source.size, precision, scale);
+                } else {
+                    parsed = !source_is_null &&
+                             CastToDecimal::from_string_lossless(source, values_to[i], precision,
+                                                                 scale, params);
+                }
+                null_map_to[i] = !parsed;
+                current_offset = next_offset;
+            }
+            block.get_by_position(result).column = std::move(nullable_col_to);
         } else {
             return Status::InternalError("Unsupported cast mode");
         }
diff --git a/be/src/exprs/function/cast/function_cast_decimal.cpp b/be/src/exprs/function/cast/function_cast_decimal.cpp
index 797bebb5f07595..cd71bf71540ac5 100644
--- a/be/src/exprs/function/cast/function_cast_decimal.cpp
+++ b/be/src/exprs/function/cast/function_cast_decimal.cpp
@@ -32,7 +32,23 @@ WrapperType create_decimal_wrapper(FunctionContext* context, const DataTypePtr&
         using Types = std::decay_t<decltype(types)>;
         using FromDataType = typename Types::LeftType;
         if constexpr (type_allow_cast_to_decimal<FromDataType>) {
-            if (context->enable_strict_mode()) {
+            if constexpr (std::is_same_v<FromDataType, DataTypeString>) {
+                if (context->enable_lossless_decimal_cast()) {
+                    if (context->enable_strict_mode()) {
+                        cast_impl = std::make_shared<CastToImpl<CastModeType::StrictLosslessMode,
+                                                                FromDataType, ToDataType>>();
+                    } else {
+                        cast_impl = std::make_shared<
+                                CastToImpl<CastModeType::LosslessMode, FromDataType, ToDataType>>();
+                    }
+                } else if (context->enable_strict_mode()) {
+                    cast_impl = std::make_shared<
+                            CastToImpl<CastModeType::StrictMode, FromDataType, ToDataType>>();
+                } else {
+                    cast_impl = std::make_shared<
+                            CastToImpl<CastModeType::NonStrictMode, FromDataType, ToDataType>>();
+                }
+            } else if (context->enable_strict_mode()) {
                 cast_impl = std::make_shared<
                         CastToImpl<CastModeType::StrictMode, FromDataType, ToDataType>>();
             } else {
diff --git a/be/src/exprs/function_context.cpp b/be/src/exprs/function_context.cpp
index 3592bcf4be8725..4284e062e1ccd5 100644
--- a/be/src/exprs/function_context.cpp
+++ b/be/src/exprs/function_context.cpp
@@ -56,6 +56,7 @@ std::unique_ptr<FunctionContext> FunctionContext::clone() {
     new_context->_fragment_local_fn_state = _fragment_local_fn_state;
     new_context->_check_overflow_for_decimal = _check_overflow_for_decimal;
     new_context->_enable_strict_mode = _enable_strict_mode;
+    new_context->_enable_lossless_decimal_cast = _enable_lossless_decimal_cast;
     new_context->_string_as_jsonb_string = _string_as_jsonb_string;
     new_context->_jsonb_string_as_string = _jsonb_string_as_string;
     return new_context;
diff --git a/be/src/exprs/function_context.h b/be/src/exprs/function_context.h
index ca0f1ed461038f..608826fff88b13 100644
--- a/be/src/exprs/function_context.h
+++ b/be/src/exprs/function_context.h
@@ -84,6 +84,8 @@ class FunctionContext {
 
     bool enable_strict_mode() const { return _enable_strict_mode; }
 
+    bool enable_lossless_decimal_cast() const { return _enable_lossless_decimal_cast; }
+
     bool set_check_overflow_for_decimal(bool check_overflow_for_decimal) {
         return _check_overflow_for_decimal = check_overflow_for_decimal;
     }
@@ -92,6 +94,10 @@ class FunctionContext {
         return _enable_strict_mode = enable_strict_mode;
     }
 
+    bool set_enable_lossless_decimal_cast(bool enable_lossless_decimal_cast) {
+        return _enable_lossless_decimal_cast = enable_lossless_decimal_cast;
+    }
+
     void set_string_as_jsonb_string(bool string_as_jsonb_string) {
         _string_as_jsonb_string = string_as_jsonb_string;
     }
@@ -198,6 +204,7 @@ class FunctionContext {
     RuntimeProfile::Counter* _udf_execute_timer = nullptr;
     bool _check_overflow_for_decimal = false;
     bool _enable_strict_mode = false;
+    bool _enable_lossless_decimal_cast = false;
 
     bool _string_as_jsonb_string = false;
     bool _jsonb_string_as_string = false;
diff --git a/be/src/exprs/vcast_expr.cpp b/be/src/exprs/vcast_expr.cpp
index 1435b40c58c58d..9435e3927c12e2 100644
--- a/be/src/exprs/vcast_expr.cpp
+++ b/be/src/exprs/vcast_expr.cpp
@@ -74,6 +74,8 @@ doris::Status VCastExpr::prepare(doris::RuntimeState* state, const doris::RowDes
                                     child->data_type()->get_name(), _target_data_type_name);
     }
     VExpr::register_function_context(state, context);
+    context->fn_context(_fn_context_index)
+            ->set_enable_lossless_decimal_cast(_lossless_decimal_cast);
     _expr_name = fmt::format("({} {}({}) TO {})", cast_name(), child_name,
                              child->data_type()->get_name(), _target_data_type_name);
     _prepare_finished = true;
diff --git a/be/src/exprs/vcast_expr.h b/be/src/exprs/vcast_expr.h
index 4ac5539ff28224..5d2196f9e56f54 100644
--- a/be/src/exprs/vcast_expr.h
+++ b/be/src/exprs/vcast_expr.h
@@ -44,7 +44,8 @@ class VCastExpr : public VExpr {
 #ifdef BE_TEST
     VCastExpr() = default;
 #endif
-    VCastExpr(const TExprNode& node) : VExpr(node) {}
+    VCastExpr(const TExprNode& node)
+            : VExpr(node), _lossless_decimal_cast(node.lossless_decimal_cast) {}
     ~VCastExpr() override = default;
     Status execute_column_impl(VExprContext* context, const Block* block, const Selector* selector,
                                size_t count, ColumnPtr& result_column) const override;
@@ -55,20 +56,24 @@ class VCastExpr : public VExpr {
     const std::string& expr_name() const override;
     std::string debug_string() const override;
     const DataTypePtr& get_target_type() const;
+    bool is_lossless_decimal_cast() const { return _lossless_decimal_cast; }
     bool is_safe_to_execute_on_selected_rows() const override { return false; }
 
     virtual std::string cast_name() const { return "CAST"; }
     Status clone_node(VExprSPtr* cloned_expr) const override {
         DORIS_CHECK(cloned_expr != nullptr);
-        *cloned_expr = VCastExpr::create_shared(clone_texpr_node());
+        auto node = clone_texpr_node();
+        node.__set_lossless_decimal_cast(_lossless_decimal_cast);
+        *cloned_expr = VCastExpr::create_shared(node);
         return Status::OK();
     }
 
     uint64_t get_digest(uint64_t seed) const override {
         auto res = VExpr::get_digest(seed);
         if (res) {
-            return HashUtil::hash64(_target_data_type_name.data(), _target_data_type_name.size(),
-                                    res);
+            res = HashUtil::hash64(_target_data_type_name.data(), _target_data_type_name.size(),
+                                   res);
+            return HashUtil::hash64(&_lossless_decimal_cast, sizeof(_lossless_decimal_cast), res);
         }
         return 0;
     }
@@ -82,6 +87,7 @@ class VCastExpr : public VExpr {
     std::string _target_data_type_name;
 
     DataTypePtr _cast_param_data_type;
+    bool _lossless_decimal_cast = false;
 
     static const constexpr char* function_name = "CAST";
 };
diff --git a/be/src/format_v2/column_mapper.cpp b/be/src/format_v2/column_mapper.cpp
index 313bbf376f860e..a5624e49273f2e 100644
--- a/be/src/format_v2/column_mapper.cpp
+++ b/be/src/format_v2/column_mapper.cpp
@@ -482,7 +482,8 @@ static Status clone_table_expr_node(const VExpr& expr, VExprSPtr* cloned_expr) {
                 split_literal->original_type(), split_literal->original_field());
     } else if (const auto* vcast_expr = dynamic_cast<const VCastExpr*>(&expr);
                vcast_expr != nullptr && vcast_expr->node_type() == TExprNodeType::CAST_EXPR) {
-        *cloned_expr = Cast::create_shared(vcast_expr->data_type());
+        *cloned_expr = Cast::create_shared(vcast_expr->data_type(),
+                                           vcast_expr->is_lossless_decimal_cast());
     }
     return Status::OK();
 }
diff --git a/be/src/format_v2/expr/cast.cpp b/be/src/format_v2/expr/cast.cpp
index efeb9d851deb22..66c662df54b08a 100644
--- a/be/src/format_v2/expr/cast.cpp
+++ b/be/src/format_v2/expr/cast.cpp
@@ -62,6 +62,8 @@ Status Cast::prepare(RuntimeState* state, const RowDescriptor& desc, VExprContex
         return Status::InternalError("Could not find function {} ", _expr_name);
     }
     VExpr::register_function_context(state, context);
+    context->fn_context(_fn_context_index)
+            ->set_enable_lossless_decimal_cast(_lossless_decimal_cast);
     _prepare_finished = true;
     return Status::OK();
 }
diff --git a/be/src/format_v2/expr/cast.h b/be/src/format_v2/expr/cast.h
index 1dc06bcf07f2bc..71878743221df3 100644
--- a/be/src/format_v2/expr/cast.h
+++ b/be/src/format_v2/expr/cast.h
@@ -38,7 +38,8 @@ class Cast final : public VExpr {
     ENABLE_FACTORY_CREATOR(Cast);
 
 public:
-    Cast(const DataTypePtr& type) {
+    Cast(const DataTypePtr& type, bool lossless_decimal_cast = false)
+            : _lossless_decimal_cast(lossless_decimal_cast) {
         _node_type = TExprNodeType::CAST_EXPR;
         _opcode = TExprOpcode::CAST;
         _data_type = type;
@@ -55,7 +56,7 @@ class Cast final : public VExpr {
     const std::string& expr_name() const override { return _expr_name; }
     Status clone_node(VExprSPtr* cloned_expr) const override {
         DORIS_CHECK(cloned_expr != nullptr);
-        *cloned_expr = Cast::create_shared(_data_type);
+        *cloned_expr = Cast::create_shared(_data_type, _lossless_decimal_cast);
         return Status::OK();
     }
 
@@ -64,5 +65,6 @@ class Cast final : public VExpr {
                        size_t count, ColumnPtr& result_column) const;
     std::string _expr_name;
     FunctionBasePtr _function;
+    bool _lossless_decimal_cast = false;
 };
 } // namespace doris::format
diff --git a/be/test/exprs/function/cast/lossless_decimal_cast_test.cpp b/be/test/exprs/function/cast/lossless_decimal_cast_test.cpp
new file mode 100644
index 00000000000000..0153ed6602247c
--- /dev/null
+++ b/be/test/exprs/function/cast/lossless_decimal_cast_test.cpp
@@ -0,0 +1,80 @@
+// 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.
+
+#include <cstring>
+#include <string>
+
+#include "exprs/function/cast/cast_to_decimal.h"
+
+namespace doris {
+
+TEST(LosslessDecimalCastTest, StringMustBeExactlyRepresentable) {
+    auto lossless = [](const char* value, UInt32 precision, UInt32 scale) {
+        return CastToDecimal::is_lossless_decimal_string(value, std::strlen(value), precision,
+                                                         scale);
+    };
+
+    EXPECT_TRUE(lossless("9223372036854775808", 38, 0));
+    EXPECT_TRUE(lossless("100.000", 38, 0));
+    EXPECT_TRUE(lossless("1.25e2", 38, 0));
+    EXPECT_TRUE(lossless("0.0100", 5, 2));
+    EXPECT_TRUE(lossless("999.99", 5, 2));
+
+    EXPECT_FALSE(lossless("100.4", 38, 0));
+    EXPECT_FALSE(lossless("1.001e2", 38, 0));
+    EXPECT_FALSE(lossless("1000", 5, 2));
+    EXPECT_FALSE(lossless("0.001", 5, 2));
+    EXPECT_FALSE(lossless("100abc", 38, 0));
+    EXPECT_FALSE(lossless("+.", 38, 0));
+    EXPECT_FALSE(lossless("999999999999999999999999999999999999999", 38, 0));
+    EXPECT_TRUE(lossless("0e9223372036854775808", 38, 0));
+    EXPECT_FALSE(lossless("0e9223372036854775808x", 38, 0));
+}
+
+TEST(LosslessDecimalCastTest, ParseOnlyExactValues) {
+    CastParameters params;
+    Decimal128V3 value;
+
+    EXPECT_TRUE(CastToDecimal::from_string_lossless(StringRef("100.000", 7), value, 38, 0, params));
+    EXPECT_EQ(value.value, 100);
+
+    EXPECT_TRUE(CastToDecimal::from_string_lossless(StringRef("1.25e2", 6), value, 38, 0, params));
+    EXPECT_EQ(value.value, 125);
+
+    EXPECT_FALSE(CastToDecimal::from_string_lossless(StringRef("100.4", 5), value, 38, 0, params));
+    EXPECT_FALSE(CastToDecimal::from_string_lossless(StringRef("100abc", 6), value, 38, 0, params));
+    EXPECT_TRUE(CastToDecimal::from_string_lossless(StringRef("0e9223372036854775808", 21), value,
+                                                    38, 0, params));
+    EXPECT_FALSE(CastToDecimal::from_string_lossless(StringRef("0e9223372036854775808x", 22), value,
+                                                     38, 0, params));
+}
+
+TEST(LosslessDecimalCastTest, HandlesCancellingExponent) {
+    std::string value = "0.";
+    value.append(1024, '0');
+    value.append("1e1025");
+
+    EXPECT_TRUE(CastToDecimal::is_lossless_decimal_string(value.data(), value.size(), 38, 0));
+
+    CastParameters params;
+    Decimal128V3 decimal;
+    EXPECT_TRUE(CastToDecimal::from_string_lossless(StringRef(value.data(), value.size()), decimal,
+                                                    38, 0, params));
+    EXPECT_EQ(decimal.value, 1);
+}
+
+} // namespace doris
diff --git a/fe/fe-catalog/src/main/java/org/apache/doris/analysis/CastExpr.java b/fe/fe-catalog/src/main/java/org/apache/doris/analysis/CastExpr.java
index 1aad243b1f2fcb..0b14adbf0d714b 100644
--- a/fe/fe-catalog/src/main/java/org/apache/doris/analysis/CastExpr.java
+++ b/fe/fe-catalog/src/main/java/org/apache/doris/analysis/CastExpr.java
@@ -33,15 +33,24 @@ public class CastExpr extends Expr {
     @SerializedName("noOp")
     protected boolean noOp = false;
 
+    // True if this implicit string-to-decimal cast must reject rounding.
+    @SerializedName("ldc")
+    protected boolean losslessDecimalCast = false;
+
     // only used restore from readFields.
     private CastExpr() {
 
     }
 
     public CastExpr(Type targetType, Expr e, boolean nullable) {
+        this(targetType, e, nullable, false);
+    }
+
+    public CastExpr(Type targetType, Expr e, boolean nullable, boolean losslessDecimalCast) {
         type = targetType;
         isImplicit = true;
         children.add(e);
+        this.losslessDecimalCast = losslessDecimalCast;
 
         noOp = Type.matchExactType(e.type, type, true);
         if (noOp) {
@@ -63,6 +72,7 @@ protected CastExpr(CastExpr other) {
         super(other);
         isImplicit = other.isImplicit;
         noOp = other.noOp;
+        losslessDecimalCast = other.losslessDecimalCast;
     }
 
     @Override
@@ -87,14 +97,19 @@ public boolean isNoOp() {
         return noOp;
     }
 
+    public boolean isLosslessDecimalCast() {
+        return losslessDecimalCast;
+    }
+
     @Override
     public int hashCode() {
-        return super.hashCode();
+        return 31 * super.hashCode() + Boolean.hashCode(losslessDecimalCast);
     }
 
     @Override
     public boolean equals(Object obj) {
-        return super.equals(obj);
+        return super.equals(obj) && obj instanceof CastExpr
+                && losslessDecimalCast == ((CastExpr) obj).losslessDecimalCast;
     }
 
     public boolean canHashPartition() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToExternalSqlVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToExternalSqlVisitor.java
index 572ae0c3508bfa..f351795b24a4dc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToExternalSqlVisitor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToExternalSqlVisitor.java
@@ -57,6 +57,10 @@ public String visitSlotRef(SlotRef expr, ToSqlParams context) {
 
     @Override
     public String visitCastExpr(CastExpr expr, ToSqlParams context) {
+        if (expr.isLosslessDecimalCast()) {
+            throw new UnsupportedOperationException(
+                    "Lossless decimal comparison casts cannot be translated to external SQL");
+        }
         return expr.getChild(0).accept(this, context);
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToThriftVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToThriftVisitor.java
index 3df252e8fed56b..67f2c2cf19c640 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToThriftVisitor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExprToThriftVisitor.java
@@ -433,6 +433,7 @@ public Void visitArithmeticExpr(ArithmeticExpr expr, TExprNode msg) {
     public Void visitCastExpr(CastExpr expr, TExprNode msg) {
         msg.node_type = TExprNodeType.CAST_EXPR;
         msg.setOpcode(TExprOpcode.CAST);
+        msg.setLosslessDecimalCast(expr.isLosslessDecimalCast());
         return null;
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExprToConnectorExpressionConverter.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExprToConnectorExpressionConverter.java
index 021782e0e53058..4056c0465474b2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExprToConnectorExpressionConverter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExprToConnectorExpressionConverter.java
@@ -106,6 +106,10 @@ public static ConnectorExpression convert(Expr expr) {
         } else if (expr instanceof LiteralExpr) {
             return convertLiteral((LiteralExpr) expr);
         } else if (expr instanceof CastExpr) {
+            if (((CastExpr) expr).isLosslessDecimalCast()) {
+                throw new UnsupportedOperationException(
+                        "Lossless decimal comparison casts cannot be pushed to connectors");
+            }
             return convert(expr.getChild(0));
         } else if (expr instanceof ArithmeticExpr) {
             return convertArithmeticExpr((ArithmeticExpr) expr);
@@ -146,6 +150,17 @@ public static ConnectorExpression convertConjuncts(List<Expr> conjuncts) {
         return new ConnectorAnd(converted);
     }
 
+    public static boolean containsLosslessDecimalCast(Expr expr) {
+        List<CastExpr> castExprs = new ArrayList<>();
+        expr.collect(CastExpr.class, castExprs);
+        return castExprs.stream().anyMatch(CastExpr::isLosslessDecimalCast);
+    }
+
+    public static boolean canPushDownLimit(List<Expr> conjuncts) {
+        return conjuncts.stream().noneMatch(
+                ExprToConnectorExpressionConverter::containsLosslessDecimalCast);
+    }
+
     /**
      * Converts a Doris {@link Type} to a {@link ConnectorType}.
      */
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/PluginDrivenScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/PluginDrivenScanNode.java
index d0875e6f32bf90..2c6254d30a0648 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/PluginDrivenScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/PluginDrivenScanNode.java
@@ -285,7 +285,13 @@ protected void convertPredicate() {
             return;
         }
         ConnectorMetadata metadata = connector.getMetadata(connectorSession);
-        ConnectorFilterConstraint constraint = buildFilterConstraint(conjuncts);
+        List<Expr> pushableConjuncts = filterPushableConjuncts(
+                metadata.supportsCastPredicatePushdown(connectorSession));
+        if (pushableConjuncts.isEmpty()) {
+            filteredToOriginalIndex = null;
+            return;
+        }
+        ConnectorFilterConstraint constraint = buildFilterConstraint(pushableConjuncts);
         Optional<FilterApplicationResult<ConnectorTableHandle>> result =
                 metadata.applyFilter(connectorSession, currentHandle, constraint);
         if (result.isPresent()) {
@@ -297,8 +303,20 @@ protected void convertPredicate() {
             // - non-null means some/all predicates remain → keep conjuncts (conservative)
             ConnectorExpression remaining = filterResult.getRemainingFilter();
             if (remaining == null) {
-                conjuncts.clear();
-                LOG.debug("Filter fully pushed down for plugin-driven scan, cleared conjuncts");
+                if (filteredToOriginalIndex == null) {
+                    conjuncts.clear();
+                } else {
+                    Set<Integer> pushedIndices = new HashSet<>(filteredToOriginalIndex);
+                    List<Expr> remainingConjuncts = new ArrayList<>();
+                    for (int i = 0; i < conjuncts.size(); i++) {
+                        if (!pushedIndices.contains(i)) {
+                            remainingConjuncts.add(conjuncts.get(i));
+                        }
+                    }
+                    conjuncts.clear();
+                    conjuncts.addAll(remainingConjuncts);
+                }
+                LOG.debug("Filter fully pushed down for plugin-driven scan");
             } else {
                 // Partial or full remaining: keep all conjuncts for BE-side evaluation.
                 // Fine-grained conjunct removal (matching individual remaining sub-expressions
@@ -320,16 +338,16 @@ protected void convertPredicate() {
      * The limit is still passed to planScan() as a parameter for
      * connectors that handle limit directly in planScan().</p>
      */
-    private void tryPushDownLimit() {
-        if (limit <= 0) {
+    private void tryPushDownLimit(long connectorLimit) {
+        if (connectorLimit <= 0) {
             return;
         }
         ConnectorMetadata metadata = connector.getMetadata(connectorSession);
         Optional<LimitApplicationResult<ConnectorTableHandle>> result =
-                metadata.applyLimit(connectorSession, currentHandle, limit);
+                metadata.applyLimit(connectorSession, currentHandle, connectorLimit);
         if (result.isPresent()) {
             currentHandle = result.get().getHandle();
-            LOG.debug("Limit {} pushed down via applyLimit for plugin-driven scan", limit);
+            LOG.debug("Limit {} pushed down via applyLimit for plugin-driven scan", connectorLimit);
         }
     }
 
@@ -354,8 +372,10 @@ private void tryPushDownProjection(List<ConnectorColumnHandle> columns) {
 
     @Override
     public List<Split> getSplits(int numBackends) throws UserException {
+        long connectorLimit = ExprToConnectorExpressionConverter.canPushDownLimit(conjuncts)
+                ? limit : -1;
         // Attempt limit and projection pushdown via SPI protocol
-        tryPushDownLimit();
+        tryPushDownLimit(connectorLimit);
 
         ConnectorScanPlanProvider scanProvider = connector.getScanPlanProvider();
         if (scanProvider == null) {
@@ -368,7 +388,7 @@ public List<Split> getSplits(int numBackends) throws UserException {
         Optional<ConnectorExpression> remainingFilter = buildRemainingFilter();
 
         List<ConnectorScanRange> ranges = scanProvider.planScan(
-                connectorSession, currentHandle, columns, remainingFilter, limit);
+                connectorSession, currentHandle, columns, remainingFilter, connectorLimit);
 
         List<Split> splits = new ArrayList<>(ranges.size());
         for (ConnectorScanRange range : ranges) {
@@ -583,30 +603,34 @@ private Optional<ConnectorExpression> buildRemainingFilter() {
             filteredToOriginalIndex = null;
             return Optional.empty();
         }
-        List<Expr> pushableConjuncts = conjuncts;
         ConnectorMetadata metadata = connector.getMetadata(connectorSession);
-        if (!metadata.supportsCastPredicatePushdown(connectorSession)) {
-            filteredToOriginalIndex = new ArrayList<>();
-            pushableConjuncts = new ArrayList<>();
-            for (int i = 0; i < conjuncts.size(); i++) {
-                if (!containsCastExpr(conjuncts.get(i))) {
-                    pushableConjuncts.add(conjuncts.get(i));
-                    filteredToOriginalIndex.add(i);
-                }
-            }
-            // If no filtering occurred, clear the mapping (1:1)
-            if (filteredToOriginalIndex.size() == conjuncts.size()) {
-                filteredToOriginalIndex = null;
-            }
-        } else {
-            filteredToOriginalIndex = null;
-        }
+        List<Expr> pushableConjuncts = filterPushableConjuncts(
+                metadata.supportsCastPredicatePushdown(connectorSession));
         if (pushableConjuncts.isEmpty()) {
             return Optional.empty();
         }
         return Optional.of(ExprToConnectorExpressionConverter.convertConjuncts(pushableConjuncts));
     }
 
+    private List<Expr> filterPushableConjuncts(boolean supportsCastPredicatePushdown) {
+        filteredToOriginalIndex = new ArrayList<>();
+        List<Expr> pushableConjuncts = new ArrayList<>();
+        for (int i = 0; i < conjuncts.size(); i++) {
+            Expr conjunct = conjuncts.get(i);
+            boolean losslessDecimalCast =
+                    ExprToConnectorExpressionConverter.containsLosslessDecimalCast(conjunct);
+            if (!losslessDecimalCast
+                    && (supportsCastPredicatePushdown || !containsCastExpr(conjunct))) {
+                pushableConjuncts.add(conjunct);
+                filteredToOriginalIndex.add(i);
+            }
+        }
+        if (filteredToOriginalIndex.size() == conjuncts.size()) {
+            filteredToOriginalIndex = null;
+        }
+        return pushableConjuncts;
+    }
+
     private static boolean containsCastExpr(Expr expr) {
         List<CastExpr> castExprs = new ArrayList<>();
         expr.collect(CastExpr.class, castExprs);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java
index 644c00028e3c97..dbf178c4a2e750 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java
@@ -32,6 +32,7 @@
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExprToConnectorExpressionConverter;
 import org.apache.doris.datasource.FileQueryScanNode;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.planner.ScanContext;
@@ -233,7 +234,7 @@ private String getQueryStr() {
             sql.append(")");
         }
 
-        if (limit != -1) {
+        if (limit != -1 && ExprToConnectorExpressionConverter.canPushDownLimit(conjuncts)) {
             sql.append(" LIMIT ").append(limit);
         }
 
@@ -241,6 +242,7 @@ private String getQueryStr() {
     }
 
     private void createFilters() {
+        filters.clear();
         if (conjuncts.isEmpty()) {
             return;
         }
@@ -257,6 +259,9 @@ private void createFilters() {
 
         ArrayList<Expr> conjunctsList = Expr.cloneList(conjuncts, sMap);
         for (Expr expr : conjunctsList) {
+            if (ExprToConnectorExpressionConverter.containsLosslessDecimalCast(expr)) {
+                continue;
+            }
             String filter = conjunctExprToString(expr, desc.getTable());
             filters.add(filter);
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java
index 84aea35f2c5001..4da339fc284ccf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java
@@ -449,7 +449,7 @@ public Expr visitCaseWhen(CaseWhen caseWhen, PlanTranslatorContext context) {
     public Expr visitCast(Cast cast, PlanTranslatorContext context) {
         // left child of cast is expression, right child of cast is target type
         CastExpr castExpr = new CastExpr(cast.getDataType().toCatalogDataType(),
-                cast.child().accept(this, context), cast.nullable());
+                cast.child().accept(this, context), cast.nullable(), cast.isLosslessDecimalCast());
         castExpr.setImplicit(!cast.isExplicitType());
         return castExpr;
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnFE.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnFE.java
index 5338bf7df9b2ef..7053e16b9c7ea6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnFE.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnFE.java
@@ -495,6 +495,9 @@ public Expression visitOr(Or or, ExpressionRewriteContext context) {
     @Override
     public Expression visitCast(Cast cast, ExpressionRewriteContext context) {
         cast = rewriteChildren(cast, context);
+        if (cast.isLosslessDecimalCast()) {
+            return cast;
+        }
         Optional<Expression> checkedExpr = preProcess(cast);
         if (checkedExpr.isPresent()) {
             return checkedExpr.get();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/SimplifyCastRule.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/SimplifyCastRule.java
index 9baabf9f58437e..9012393d4adafb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/SimplifyCastRule.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/SimplifyCastRule.java
@@ -69,6 +69,10 @@ public List<ExpressionPatternMatcher<? extends Expression>> buildRules() {
     public static Expression simplifyCast(Cast cast) {
         Expression child = cast.child();
 
+        if (cast.isLosslessDecimalCast()) {
+            return cast;
+        }
+
         // remove redundant cast
         // CAST(value as type), value is type
         if (cast.getDataType().equals(child.getDataType())) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/LogicalResultSinkToShortCircuitPointQuery.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/LogicalResultSinkToShortCircuitPointQuery.java
index dfcd2ea289cc6f..a71e366072f3d7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/LogicalResultSinkToShortCircuitPointQuery.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/LogicalResultSinkToShortCircuitPointQuery.java
@@ -50,10 +50,16 @@ private Expression removeCast(Expression expression) {
         return expression;
     }
 
+    private boolean isLosslessDecimalCast(Expression expression) {
+        return expression instanceof Cast && ((Cast) expression).isLosslessDecimalCast();
+    }
+
     private boolean filterMatchShortCircuitCondition(LogicalFilter<LogicalOlapScan> filter) {
         return filter.getConjuncts().stream().allMatch(
                 // all conjuncts match with pattern `key = ?`
                 expression -> (expression instanceof EqualTo)
+                        && !isLosslessDecimalCast(expression.child(0))
+                        && !isLosslessDecimalCast(expression.child(1))
                         && (removeCast(expression.child(0)).isKeyColumnFromTable()
                         || (expression.child(0) instanceof SlotReference
                         && ((SlotReference) expression.child(0)).getName().equals(Column.DELETE_SIGN)))
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SlotTypeReplacer.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SlotTypeReplacer.java
index 6a8fd28b902ba7..bf4f5f7e1a5482 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SlotTypeReplacer.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SlotTypeReplacer.java
@@ -618,7 +618,7 @@ private Expression rewriteCast(Cast cast, boolean fillAccessPath) {
             newType = prunedTree.pruneCastType(originTree, castTree);
         }
 
-        return new Cast(newChild, newType);
+        return new Cast(newChild, newType, cast.isExplicitType(), cast.isLosslessDecimalCast());
     }
 
     private List<ColumnAccessPath> replaceIcebergAccessPathToId(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Cast.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Cast.java
index 1e14d3284ea768..af539ad01abfae 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Cast.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Cast.java
@@ -46,6 +46,9 @@ public class Cast extends Expression implements UnaryExpression, Monotonic {
     // CAST can be from SQL Query or Type Coercion. true for explicitly cast from SQL query.
     protected final boolean isExplicitType; //FIXME: now not useful
 
+    // Only used by implicit numeric-string equality comparisons.
+    protected final boolean losslessDecimalCast;
+
     protected final DataType targetType;
 
     public Cast(Expression child, DataType targetType) {
@@ -53,19 +56,29 @@ public Cast(Expression child, DataType targetType) {
     }
 
     public Cast(Expression child, DataType targetType, boolean isExplicitType) {
-        this(ImmutableList.of(child), targetType, isExplicitType);
+        this(child, targetType, isExplicitType, false);
+    }
+
+    public Cast(Expression child, DataType targetType, boolean isExplicitType, boolean losslessDecimalCast) {
+        this(ImmutableList.of(child), targetType, isExplicitType, losslessDecimalCast);
     }
 
-    protected Cast(List<Expression> child, DataType targetType, boolean isExplicitType) {
+    protected Cast(List<Expression> child, DataType targetType, boolean isExplicitType,
+            boolean losslessDecimalCast) {
         super(child);
         this.targetType = Objects.requireNonNull(targetType, "targetType can not be null");
         this.isExplicitType = isExplicitType;
+        this.losslessDecimalCast = losslessDecimalCast;
     }
 
     public boolean isExplicitType() {
         return isExplicitType;
     }
 
+    public boolean isLosslessDecimalCast() {
+        return losslessDecimalCast;
+    }
+
     @Override
     public DataType getDataType() {
         return targetType;
@@ -213,7 +226,7 @@ public static boolean castNullable(boolean srcNullable, DataType srcType, DataTy
     @Override
     public Cast withChildren(List<Expression> children) {
         Preconditions.checkArgument(children.size() == 1);
-        return new Cast(children, targetType, isExplicitType);
+        return new Cast(children, targetType, isExplicitType, losslessDecimalCast);
     }
 
     @Override
@@ -248,12 +261,13 @@ public boolean equals(Object o) {
             return false;
         }
         Cast cast = (Cast) o;
-        return Objects.equals(targetType, cast.targetType);
+        return Objects.equals(targetType, cast.targetType)
+                && losslessDecimalCast == cast.losslessDecimalCast;
     }
 
     @Override
     public int computeHashCode() {
-        return Objects.hash(super.computeHashCode(), targetType);
+        return Objects.hash(super.computeHashCode(), targetType, losslessDecimalCast);
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/TryCast.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/TryCast.java
index 2398d7d49e3e0b..6314e3b8271d35 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/TryCast.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/TryCast.java
@@ -44,7 +44,7 @@ public TryCast(Expression child, DataType targetType, boolean isExplicitType) {
     }
 
     private TryCast(List<Expression> child, DataType targetType, boolean isExplicitType) {
-        super(child, targetType, isExplicitType);
+        super(child, targetType, isExplicitType, false);
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CopyIntoInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CopyIntoInfo.java
index 6fd99573b805b5..8aad2f56bddfe2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CopyIntoInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CopyIntoInfo.java
@@ -341,9 +341,10 @@ public Expr visitSlotReference(SlotReference slotReference, PlanTranslatorContex
 
         @Override
         public Expr visitCast(Cast cast, PlanTranslatorContext context) {
-            // left child of cast is target type, right child of cast is expression
-            return new CastExpr(cast.getDataType().toCatalogDataType(),
-                    cast.child().accept(this, context), cast.nullable());
+            CastExpr legacyCast = new CastExpr(cast.getDataType().toCatalogDataType(),
+                    cast.child().accept(this, context), cast.nullable(), cast.isLosslessDecimalCast());
+            legacyCast.setImplicit(!cast.isExplicitType());
+            return legacyCast;
         }
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
index c19250e4060f2f..c826aad4d11fd9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
@@ -35,6 +35,7 @@
 import org.apache.doris.nereids.trees.expressions.Cast;
 import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
 import org.apache.doris.nereids.trees.expressions.Divide;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.InPredicate;
 import org.apache.doris.nereids.trees.expressions.IntegralDivide;
@@ -474,6 +475,42 @@ public static Expression castIfNotSameType(Expression input, DataType targetType
         }
     }
 
+    private static Expression castComparisonOperand(
+            Expression input, DataType targetType, boolean losslessDecimalCast) {
+        if (losslessDecimalCast && input.getDataType().isStringLikeType()
+                && targetType.isDecimalV3Type()) {
+            checkCanCastTo(input.getDataType(), targetType);
+            return new Cast(input, targetType, false, true);
+        }
+        return castIfNotSameType(input, targetType);
+    }
+
+    private static Optional<DataType> getLosslessDecimalStringComparisonType(
+            ComparisonPredicate comparisonPredicate, Expression left, Expression right) {
+        if (!(comparisonPredicate instanceof EqualTo)) {
+            return Optional.empty();
+        }
+
+        DataType decimalType;
+        if (left.getDataType().isDecimalLikeType() && right.getDataType().isStringLikeType()) {
+            decimalType = left.getDataType();
+        } else if (right.getDataType().isDecimalLikeType() && left.getDataType().isStringLikeType()) {
+            decimalType = right.getDataType();
+        } else {
+            return Optional.empty();
+        }
+
+        DecimalV3Type decimalV3Type = DecimalV3Type.forType(decimalType);
+        int maxPrecision = SessionVariable.getEnableDecimal256()
+                ? DecimalV3Type.MAX_DECIMAL256_PRECISION : DecimalV3Type.MAX_DECIMAL128_PRECISION;
+        int integerPart = Math.max(decimalV3Type.getPrecision() - decimalV3Type.getScale(), 0);
+        int maxScale = Math.max(maxPrecision - integerPart, 0);
+        int targetScale = Math.max(decimalV3Type.getScale(),
+                Math.min(SessionVariable.getDecimalOverFlowScale(), maxScale));
+        targetScale = Math.min(targetScale, maxScale);
+        return Optional.of(DecimalV3Type.createDecimalV3Type(maxPrecision, targetScale));
+    }
+
     /**
      * Wrap {@code expression} in a cast to {@code targetType} when the source type can
      * already be resolved (Literal, or any expression whose {@code getDataType()} does
@@ -1299,22 +1336,25 @@ public static Expression processComparisonPredicate(ComparisonPredicate comparis
         left = comparisonPredicate.left();
         right = comparisonPredicate.right();
 
-        Optional<DataType> commonType;
-        if (GlobalVariable.enableNewTypeCoercionBehavior) {
-            commonType = findWiderTypeForTwo(left.getDataType(), right.getDataType(), false, false);
-        } else {
-            commonType = findWiderTypeForTwoForComparison(left.getDataType(), right.getDataType(), false);
-        }
+        Optional<DataType> commonType = getLosslessDecimalStringComparisonType(comparisonPredicate, left, right);
+        boolean losslessDecimalCast = commonType.isPresent();
+        if (!losslessDecimalCast) {
+            if (GlobalVariable.enableNewTypeCoercionBehavior) {
+                commonType = findWiderTypeForTwo(left.getDataType(), right.getDataType(), false, false);
+            } else {
+                commonType = findWiderTypeForTwoForComparison(left.getDataType(), right.getDataType(), false);
+            }
 
-        if (commonType.isPresent()) {
-            commonType = Optional.of(downgradeDecimalAndDateLikeType(
-                    commonType.get(),
-                    left,
-                    right));
-            commonType = Optional.of(downgradeDecimalAndDateLikeType(
-                    commonType.get(),
-                    right,
-                    left));
+            if (commonType.isPresent()) {
+                commonType = Optional.of(downgradeDecimalAndDateLikeType(
+                        commonType.get(),
+                        left,
+                        right));
+                commonType = Optional.of(downgradeDecimalAndDateLikeType(
+                        commonType.get(),
+                        right,
+                        left));
+            }
         }
 
         if (commonType.isPresent()) {
@@ -1322,8 +1362,8 @@ public static Expression processComparisonPredicate(ComparisonPredicate comparis
                 throw new AnalysisException("data type " + commonType.get()
                         + " could not used in ComparisonPredicate " + comparisonPredicate.toSql());
             }
-            left = castIfNotSameType(left, commonType.get());
-            right = castIfNotSameType(right, commonType.get());
+            left = castComparisonOperand(left, commonType.get(), losslessDecimalCast);
+            right = castComparisonOperand(right, commonType.get(), losslessDecimalCast);
         } else {
             throw new AnalysisException("unsupported comparison predicate " + comparisonPredicate.toSql());
         }
diff --git a/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToSqlTest.java b/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToSqlTest.java
index 24fceb2c901259..b80166c7d3fad4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToSqlTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToSqlTest.java
@@ -26,6 +26,7 @@
 import org.apache.doris.catalog.Type;
 import org.apache.doris.catalog.info.TableNameInfo;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.ExprToConnectorExpressionConverter;
 import org.apache.doris.nereids.util.MoreFieldsThread;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
@@ -597,6 +598,19 @@ public void testCastExprExternal() {
         Assertions.assertEquals("1", sql);
     }
 
+    @Test
+    public void testLosslessDecimalCastIsNotPushedDown() {
+        CastExpr expr = new CastExpr(ScalarType.createDecimalV3Type(38, 0),
+                new StringLiteral("100.4"), true, true);
+        Assertions.assertThrows(UnsupportedOperationException.class,
+                () -> expr.accept(ExprToExternalSqlVisitor.INSTANCE,
+                        new ToSqlParams(false, true, null, null)));
+        Assertions.assertThrows(UnsupportedOperationException.class,
+                () -> ExprToConnectorExpressionConverter.convert(expr));
+        Assertions.assertFalse(ExprToConnectorExpressionConverter.canPushDownLimit(
+                Collections.singletonList(expr)));
+    }
+
     @Test
     public void testTryCastExprDefault() {
         TryCastExpr expr = new TryCastExpr(Type.INT, new IntLiteral(1L), false, false);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToThriftBehaviorTest.java b/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToThriftBehaviorTest.java
index ac40a4f174f682..91bb7ab597dd79 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToThriftBehaviorTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/analysis/ExprToThriftBehaviorTest.java
@@ -258,6 +258,16 @@ public void testCastExprNoOpTrue() {
         Assertions.assertEquals(TExprNodeType.INT_LITERAL, nodes.get(0).node_type);
     }
 
+    @Test
+    public void testLosslessDecimalCastExpr() {
+        IntLiteral child = new IntLiteral(42);
+        CastExpr expr = new CastExpr(Type.BIGINT, child, false, true);
+
+        TExprNode node = firstNode(expr);
+        Assertions.assertTrue(node.isSetLosslessDecimalCast());
+        Assertions.assertTrue(node.isLosslessDecimalCast());
+    }
+
     // ======================== treesToThrift ========================
 
     @Test
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/util/TypeCoercionUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/TypeCoercionUtilsTest.java
index d2ead4e326d667..fc79eb60402d7d 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/util/TypeCoercionUtilsTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/TypeCoercionUtilsTest.java
@@ -23,8 +23,10 @@
 import org.apache.doris.nereids.trees.expressions.Divide;
 import org.apache.doris.nereids.trees.expressions.EqualTo;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
 import org.apache.doris.nereids.trees.expressions.InPredicate;
 import org.apache.doris.nereids.trees.expressions.Multiply;
+import org.apache.doris.nereids.trees.expressions.NullSafeEqual;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.expressions.Subtract;
 import org.apache.doris.nereids.trees.expressions.functions.BoundFunction;
@@ -69,6 +71,7 @@
 import org.apache.doris.nereids.types.VarcharType;
 import org.apache.doris.nereids.types.coercion.IntegralType;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.GlobalVariable;
 
 import com.google.common.collect.ImmutableList;
 import org.junit.jupiter.api.Assertions;
@@ -250,6 +253,69 @@ public void testProcessComparisonPredicateDowngrade() {
         Assertions.assertEquals(DateTimeType.INSTANCE, datetimeDowngrade.left().getDataType());
     }
 
+    @Test
+    public void testLosslessDecimalStringComparisonCoercion() {
+        boolean originalNewTypeCoercionBehavior = GlobalVariable.enableNewTypeCoercionBehavior;
+        try {
+            for (boolean newTypeCoercionBehavior : new boolean[] {true, false}) {
+                GlobalVariable.enableNewTypeCoercionBehavior = newTypeCoercionBehavior;
+                DecimalV3Type decimal380 = DecimalV3Type.createDecimalV3Type(38, 0);
+                EqualTo decimalString = new EqualTo(
+                        new SlotReference("decimal_col", decimal380),
+                        new SlotReference("string_col", StringType.INSTANCE));
+                decimalString = (EqualTo) TypeCoercionUtils.processComparisonPredicate(decimalString);
+                Assertions.assertEquals(decimal380, decimalString.left().getDataType());
+                Cast decimalStringCast = (Cast) decimalString.right();
+                Assertions.assertEquals(decimal380, decimalStringCast.getDataType());
+                Assertions.assertTrue(decimalStringCast.isLosslessDecimalCast());
+
+                DecimalV3Type decimal54 = DecimalV3Type.createDecimalV3Type(5, 4);
+                EqualTo decimalChar = new EqualTo(
+                        new SlotReference("decimal_col", decimal54),
+                        new SlotReference("char_col", CharType.createCharType(255)));
+                decimalChar = (EqualTo) TypeCoercionUtils.processComparisonPredicate(decimalChar);
+                DecimalV3Type decimal386 = DecimalV3Type.createDecimalV3Type(38, 6);
+                Assertions.assertEquals(decimal386, decimalChar.left().getDataType());
+                Assertions.assertEquals(decimal386, decimalChar.right().getDataType());
+                Assertions.assertTrue(((Cast) decimalChar.right()).isLosslessDecimalCast());
+
+                NullSafeEqual nullSafeDecimalString = new NullSafeEqual(
+                        new SlotReference("decimal_col", decimal380),
+                        new SlotReference("string_col", StringType.INSTANCE));
+                nullSafeDecimalString = (NullSafeEqual) TypeCoercionUtils
+                        .processComparisonPredicate(nullSafeDecimalString);
+                Assertions.assertFalse(((Cast) nullSafeDecimalString.right()).isLosslessDecimalCast());
+
+                EqualTo bigintString = new EqualTo(
+                        new SlotReference("bigint_col", BigIntType.INSTANCE),
+                        new SlotReference("string_col", StringType.INSTANCE));
+                bigintString = (EqualTo) TypeCoercionUtils.processComparisonPredicate(bigintString);
+                Assertions.assertFalse(((Cast) bigintString.right()).isLosslessDecimalCast());
+
+                EqualTo floatString = new EqualTo(
+                        new SlotReference("float_col", FloatType.INSTANCE),
+                        new SlotReference("string_col", StringType.INSTANCE));
+                floatString = (EqualTo) TypeCoercionUtils.processComparisonPredicate(floatString);
+                Assertions.assertEquals(DoubleType.INSTANCE, floatString.left().getDataType());
+                Assertions.assertFalse(((Cast) floatString.right()).isLosslessDecimalCast());
+
+                GreaterThan rangeComparison = new GreaterThan(
+                        new SlotReference("decimal_col", decimal380),
+                        new SlotReference("string_col", StringType.INSTANCE));
+                rangeComparison = (GreaterThan) TypeCoercionUtils.processComparisonPredicate(rangeComparison);
+                Assertions.assertFalse(((Cast) rangeComparison.right()).isLosslessDecimalCast());
+
+                InPredicate inPredicate = new InPredicate(
+                        new SlotReference("decimal_col", decimal380),
+                        ImmutableList.of(new SlotReference("string_col", StringType.INSTANCE)));
+                inPredicate = (InPredicate) TypeCoercionUtils.processInPredicate(inPredicate);
+                Assertions.assertFalse(((Cast) inPredicate.getOptions().get(0)).isLosslessDecimalCast());
+            }
+        } finally {
+            GlobalVariable.enableNewTypeCoercionBehavior = originalNewTypeCoercionBehavior;
+        }
+    }
+
     @Test
     public void testProcessInStringCoercion() {
         // BigInt slot vs String literal
diff --git a/gensrc/thrift/Exprs.thrift b/gensrc/thrift/Exprs.thrift
index 812a65740a77c2..653c1b56fffc83 100644
--- a/gensrc/thrift/Exprs.thrift
+++ b/gensrc/thrift/Exprs.thrift
@@ -333,6 +333,8 @@ struct TExprNode {
   // distinguish current-scope lambda arguments from captured outer lambda
   // arguments when nested lambda expressions contain duplicated column ids.
   42: optional list<string> lambda_argument_names
+  // Use lossless parsing for implicit string-to-decimal equality comparisons.
+  43: optional bool lossless_decimal_cast
 }
 
 // A flattened representation of a tree of Expr nodes, obtained by depth-first
diff --git a/regression-test/data/nereids_rules_p0/infer_predicate/pull_up_predicate_literal.out b/regression-test/data/nereids_rules_p0/infer_predicate/pull_up_predicate_literal.out
index e352e38da3ae18..4eb65b96a23de7 100644
--- a/regression-test/data/nereids_rules_p0/infer_predicate/pull_up_predicate_literal.out
+++ b/regression-test/data/nereids_rules_p0/infer_predicate/pull_up_predicate_literal.out
@@ -1137,7 +1137,7 @@ PhysicalResultSink
 ----------PhysicalProject
 ------------PhysicalStorageLayerAggregate[test_pull_up_predicate_literal]
 ------PhysicalProject
---------filter((cast(t2.d_char as DECIMALV3(38, 6)) = 1.000000))
+--------filter((cast(t2.d_char as DECIMALV3(38, 12)) = 1.000000000003))
 ----------PhysicalOlapScan[test_types(t2)]
 
 -- !const_value_and_join_column_type124 --
@@ -1149,7 +1149,7 @@ PhysicalResultSink
 ----------PhysicalProject
 ------------PhysicalStorageLayerAggregate[test_pull_up_predicate_literal]
 ------PhysicalProject
---------filter((cast(t2.d_varchar as DECIMALV3(38, 6)) = 1.000000))
+--------filter((cast(t2.d_varchar as DECIMALV3(38, 12)) = 1.000000000003))
 ----------PhysicalOlapScan[test_types(t2)]
 
 -- !const_value_and_join_column_type125 --
@@ -1161,7 +1161,7 @@ PhysicalResultSink
 ----------PhysicalProject
 ------------PhysicalStorageLayerAggregate[test_pull_up_predicate_literal]
 ------PhysicalProject
---------filter((cast(t2.d_string as DECIMALV3(38, 6)) = 1.000000))
+--------filter((cast(t2.d_string as DECIMALV3(38, 12)) = 1.000000000003))
 ----------PhysicalOlapScan[test_types(t2)]
 
 -- !const_value_and_join_column_type128 --
@@ -1257,7 +1257,7 @@ PhysicalResultSink
 ----------PhysicalProject
 ------------PhysicalStorageLayerAggregate[test_pull_up_predicate_literal]
 ------PhysicalProject
---------filter((cast(t2.d_char as DECIMALV3(38, 6)) = 12232.239827))
+--------filter((cast(t2.d_char as DECIMALV3(38, 16)) = 12232.2398272342335234))
 ----------PhysicalOlapScan[test_types(t2)]
 
 -- !const_value_and_join_column_type140 --
@@ -1269,7 +1269,7 @@ PhysicalResultSink
 ----------PhysicalProject
 ------------PhysicalStorageLayerAggregate[test_pull_up_predicate_literal]
 ------PhysicalProject
---------filter((cast(t2.d_varchar as DECIMALV3(38, 6)) = 12232.239827))
+--------filter((cast(t2.d_varchar as DECIMALV3(38, 16)) = 12232.2398272342335234))
 ----------PhysicalOlapScan[test_types(t2)]
 
 -- !const_value_and_join_column_type141 --
@@ -1281,7 +1281,7 @@ PhysicalResultSink
 ----------PhysicalProject
 ------------PhysicalStorageLayerAggregate[test_pull_up_predicate_literal]
 ------PhysicalProject
---------filter((cast(t2.d_string as DECIMALV3(38, 6)) = 12232.239827))
+--------filter((cast(t2.d_string as DECIMALV3(38, 16)) = 12232.2398272342335234))
 ----------PhysicalOlapScan[test_types(t2)]
 
 -- !const_value_and_join_column_type144 --
diff --git a/regression-test/suites/correctness_p0/test_join_decimal_string_lossless_cast.groovy b/regression-test/suites/correctness_p0/test_join_decimal_string_lossless_cast.groovy
new file mode 100644
index 00000000000000..58485302d42bbd
--- /dev/null
+++ b/regression-test/suites/correctness_p0/test_join_decimal_string_lossless_cast.groovy
@@ -0,0 +1,111 @@
+// 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_join_decimal_string_lossless_cast") {
+    def originStrictCast = sql("select @@enable_strict_cast")
+
+    sql """drop table if exists test_join_decimal_string_numeric"""
+    sql """drop table if exists test_join_decimal_string_value"""
+
+    sql """
+        create table test_join_decimal_string_numeric (
+            id int,
+            k decimal(38, 0)
+        ) engine=olap
+        duplicate key(id)
+        distributed by hash(id) buckets 1
+        properties ("replication_allocation" = "tag.location.default: 1")
+    """
+    sql """
+        create table test_join_decimal_string_value (
+            k varchar(64)
+        ) engine=olap
+        duplicate key(k)
+        distributed by hash(k) buckets 1
+        properties ("replication_allocation" = "tag.location.default: 1")
+    """
+
+    sql """
+        insert into test_join_decimal_string_numeric values
+            (1, 100),
+            (2, 9007199254740992),
+            (3, 9007199254740993),
+            (4, 99999999999999999999999999999999999999),
+            (5, 0)
+    """
+    sql """
+        insert into test_join_decimal_string_value values
+            ('100'),
+            ('100.000'),
+            ('100.4'),
+            ('100.5'),
+            ('100abc'),
+            ('9007199254740992'),
+            ('9007199254740993'),
+            ('99999999999999999999999999999999999999'),
+            ('999999999999999999999999999999999999999'),
+            ('0e9223372036854775808'),
+            ('0e9223372036854775808x')
+    """
+
+    sql """set enable_strict_cast = false"""
+
+    test {
+        sql """
+            select count(*)
+            from test_join_decimal_string_numeric n
+            join test_join_decimal_string_value s on n.k = s.k
+            where n.id = 1
+        """
+        result([[2L]])
+    }
+
+    test {
+        sql """
+            select count(*)
+            from test_join_decimal_string_numeric n
+            join test_join_decimal_string_value s on n.k = s.k
+        """
+        result([[6L]])
+    }
+
+    sql """set enable_strict_cast = true"""
+
+    test {
+        sql """
+            select count(*)
+            from test_join_decimal_string_numeric n
+            join test_join_decimal_string_value s on n.k = s.k
+            where n.id = 1 and s.k = '100abc'
+        """
+        exception "parse number fail"
+    }
+
+    test {
+        sql """
+            select count(*)
+            from test_join_decimal_string_numeric n
+            join test_join_decimal_string_value s on n.k = s.k
+            where n.id = 4 and s.k = '999999999999999999999999999999999999999'
+        """
+        exception "parse number fail"
+    }
+
+    sql """set enable_strict_cast = ${originStrictCast[0][0]}"""
+    sql """drop table if exists test_join_decimal_string_numeric"""
+    sql """drop table if exists test_join_decimal_string_value"""
+}