diff --git a/be/src/gutil/endian.h b/be/src/gutil/endian.h
index 4434bb943b47d8..16df153961f496 100644
--- a/be/src/gutil/endian.h
+++ b/be/src/gutil/endian.h
@@ -35,6 +35,7 @@
 #include "gutil/int128.h"
 #include "gutil/integral_types.h"
 #include "gutil/port.h"
+#include "vec/core/wide_integer.h"
 
 inline uint64 gbswap_64(uint64 host_int) {
 #if defined(__GNUC__) && defined(__x86_64__) && !defined(__APPLE__)
@@ -59,6 +60,11 @@ inline unsigned __int128 gbswap_128(unsigned __int128 host_int) {
            (static_cast<unsigned __int128>(bswap_64(static_cast<uint64>(host_int))) << 64);
 }
 
+inline wide::UInt256 gbswap_256(wide::UInt256 host_int) {
+    wide::UInt256 result{gbswap_64(host_int.items[0]), gbswap_64(host_int.items[1]), gbswap_64(host_int.items[2]), gbswap_64(host_int.items[3])};
+    return result;
+}
+
 // Swap bytes of a 24-bit value.
 inline uint32_t bswap_24(uint32_t x) {
     return ((x & 0x0000ffULL) << 16) | ((x & 0x00ff00ULL)) | ((x & 0xff0000ULL) >> 16);
@@ -252,6 +258,9 @@ class BigEndian {
     static unsigned __int128 FromHost128(unsigned __int128 x) { return gbswap_128(x); }
     static unsigned __int128 ToHost128(unsigned __int128 x) { return gbswap_128(x); }
 
+    static wide::UInt256 FromHost256(wide::UInt256 x) { return gbswap_256(x); }
+    static wide::UInt256 ToHost256(wide::UInt256 x) { return gbswap_256(x); }
+
     static bool IsLittleEndian() { return true; }
 
 #elif defined IS_BIG_ENDIAN
@@ -271,6 +280,9 @@ class BigEndian {
     static uint128 FromHost128(uint128 x) { return x; }
     static uint128 ToHost128(uint128 x) { return x; }
 
+    static wide::UInt256 FromHost128(wide::UInt256 x) { return x; }
+    static wide::UInt256 ToHost128(wide::UInt256 x) { return x; }
+
     static bool IsLittleEndian() { return false; }
 
 #endif /* ENDIAN */
diff --git a/be/src/olap/delete_handler.cpp b/be/src/olap/delete_handler.cpp
index df0f6211aa3a24..dc3aab6acbbcc7 100644
--- a/be/src/olap/delete_handler.cpp
+++ b/be/src/olap/delete_handler.cpp
@@ -199,6 +199,8 @@ bool DeleteHandler::is_condition_value_valid(const TabletColumn& column,
         return valid_decimal(value_str, column.precision(), column.frac());
     case FieldType::OLAP_FIELD_TYPE_DECIMAL128I:
         return valid_decimal(value_str, column.precision(), column.frac());
+    case FieldType::OLAP_FIELD_TYPE_DECIMAL256:
+        return valid_decimal(value_str, column.precision(), column.frac());
     case FieldType::OLAP_FIELD_TYPE_CHAR:
     case FieldType::OLAP_FIELD_TYPE_VARCHAR:
         return value_str.size() <= column.length();
diff --git a/be/src/olap/field.h b/be/src/olap/field.h
index fc9c87bfb1c3ae..a92a8f121c149f 100644
--- a/be/src/olap/field.h
+++ b/be/src/olap/field.h
@@ -589,6 +589,8 @@ class FieldFactory {
                 [[fallthrough]];
             case FieldType::OLAP_FIELD_TYPE_DECIMAL128I:
                 [[fallthrough]];
+            case FieldType::OLAP_FIELD_TYPE_DECIMAL256:
+                [[fallthrough]];
             case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
                 Field* field = new Field(column);
                 field->set_precision(column.precision());
@@ -647,6 +649,8 @@ class FieldFactory {
                 [[fallthrough]];
             case FieldType::OLAP_FIELD_TYPE_DECIMAL128I:
                 [[fallthrough]];
+            case FieldType::OLAP_FIELD_TYPE_DECIMAL256:
+                [[fallthrough]];
             case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
                 Field* field = new Field(column);
                 field->set_precision(column.precision());
diff --git a/be/src/olap/key_coder.cpp b/be/src/olap/key_coder.cpp
index 803b353375d2d9..168117117d9116 100644
--- a/be/src/olap/key_coder.cpp
+++ b/be/src/olap/key_coder.cpp
@@ -80,6 +80,7 @@ class KeyCoderResolver {
         add_mapping<FieldType::OLAP_FIELD_TYPE_DECIMAL32>();
         add_mapping<FieldType::OLAP_FIELD_TYPE_DECIMAL64>();
         add_mapping<FieldType::OLAP_FIELD_TYPE_DECIMAL128I>();
+        add_mapping<FieldType::OLAP_FIELD_TYPE_DECIMAL256>();
     }
 
     template <FieldType field_type>
diff --git a/be/src/olap/key_coder.h b/be/src/olap/key_coder.h
index 30d33cd3faec00..6885a0d96f251b 100644
--- a/be/src/olap/key_coder.h
+++ b/be/src/olap/key_coder.h
@@ -85,6 +85,7 @@ class KeyCoderTraits<
         field_type,
         typename std::enable_if<
                 std::is_integral<typename CppTypeTraits<field_type>::CppType>::value ||
+                field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL256 ||
                 vectorized::IsDecimalNumber<typename CppTypeTraits<field_type>::CppType>>::type> {
 public:
     using CppType = typename CppTypeTraits<field_type>::CppType;
@@ -93,20 +94,24 @@ class KeyCoderTraits<
 private:
     // Swap value's endian from/to big endian
     static UnsignedCppType swap_big_endian(UnsignedCppType val) {
-        switch (sizeof(UnsignedCppType)) {
-        case 1:
-            return val;
-        case 2:
-            return BigEndian::FromHost16(val);
-        case 4:
-            return BigEndian::FromHost32(val);
-        case 8:
-            return BigEndian::FromHost64(val);
-        case 16:
-            return BigEndian::FromHost128(val);
-        default:
-            LOG(FATAL) << "Invalid type to big endian, type=" << int(field_type)
-                       << ", size=" << sizeof(UnsignedCppType);
+        if constexpr (field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL256) {
+            return BigEndian::FromHost256(val);
+        } else {
+            switch (sizeof(UnsignedCppType)) {
+            case 1:
+                return val;
+            case 2:
+                return BigEndian::FromHost16(val);
+            case 4:
+                return BigEndian::FromHost32(val);
+            case 8:
+                return BigEndian::FromHost64(val);
+            case 16:
+                return BigEndian::FromHost128(val);
+            default:
+                LOG(FATAL) << "Invalid type to big endian, type=" << int(field_type)
+                           << ", size=" << sizeof(UnsignedCppType);
+            }
         }
     }
 
diff --git a/be/src/olap/olap_common.h b/be/src/olap/olap_common.h
index b2a6ac49cd1c0b..3811aab378f6d8 100644
--- a/be/src/olap/olap_common.h
+++ b/be/src/olap/olap_common.h
@@ -198,6 +198,7 @@ constexpr bool field_is_numeric_type(const FieldType& field_type) {
            field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL32 ||
            field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL64 ||
            field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL128I ||
+           field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL256 ||
            field_type == FieldType::OLAP_FIELD_TYPE_BOOL;
 }
 
diff --git a/be/src/olap/rowset/segment_v2/bitshuffle_page.h b/be/src/olap/rowset/segment_v2/bitshuffle_page.h
index 05d07acf88462b..54f446070f106f 100644
--- a/be/src/olap/rowset/segment_v2/bitshuffle_page.h
+++ b/be/src/olap/rowset/segment_v2/bitshuffle_page.h
@@ -267,6 +267,7 @@ inline Status parse_bit_shuffle_header(const Slice& data, size_t& num_elements,
     case 8:
     case 12:
     case 16:
+    case 32:
         break;
     default:
         return Status::InternalError("invalid size_of_elem:{}", size_of_element);
diff --git a/be/src/olap/rowset/segment_v2/encoding_info.cpp b/be/src/olap/rowset/segment_v2/encoding_info.cpp
index 573ea925325a98..462b5bdf51c2a4 100644
--- a/be/src/olap/rowset/segment_v2/encoding_info.cpp
+++ b/be/src/olap/rowset/segment_v2/encoding_info.cpp
@@ -321,6 +321,10 @@ EncodingInfoResolver::EncodingInfoResolver() {
     _add_map<FieldType::OLAP_FIELD_TYPE_DECIMAL128I, PLAIN_ENCODING>();
     _add_map<FieldType::OLAP_FIELD_TYPE_DECIMAL128I, BIT_SHUFFLE, true>();
 
+    _add_map<FieldType::OLAP_FIELD_TYPE_DECIMAL256, BIT_SHUFFLE>();
+    _add_map<FieldType::OLAP_FIELD_TYPE_DECIMAL256, PLAIN_ENCODING>();
+    _add_map<FieldType::OLAP_FIELD_TYPE_DECIMAL256, BIT_SHUFFLE, true>();
+
     _add_map<FieldType::OLAP_FIELD_TYPE_HLL, PLAIN_ENCODING>();
 
     _add_map<FieldType::OLAP_FIELD_TYPE_OBJECT, PLAIN_ENCODING>();
diff --git a/be/src/olap/rowset/segment_v2/zone_map_index.cpp b/be/src/olap/rowset/segment_v2/zone_map_index.cpp
index 9368862fff6ac2..69c1812d14148a 100644
--- a/be/src/olap/rowset/segment_v2/zone_map_index.cpp
+++ b/be/src/olap/rowset/segment_v2/zone_map_index.cpp
@@ -198,7 +198,8 @@ Status ZoneMapIndexReader::_load(bool use_page_cache, bool kept_in_memory,
     M(TYPE_STRING)               \
     M(TYPE_DECIMAL32)            \
     M(TYPE_DECIMAL64)            \
-    M(TYPE_DECIMAL128I)
+    M(TYPE_DECIMAL128I)          \
+    M(TYPE_DECIMAL256)
 
 Status ZoneMapIndexWriter::create(Field* field, std::unique_ptr<ZoneMapIndexWriter>& res) {
     switch (field->type()) {
diff --git a/be/src/olap/schema.cpp b/be/src/olap/schema.cpp
index d465db4f1d0fb4..142e56c55cb200 100644
--- a/be/src/olap/schema.cpp
+++ b/be/src/olap/schema.cpp
@@ -199,6 +199,9 @@ vectorized::IColumn::MutablePtr Schema::get_predicate_column_ptr(const Field& fi
     case FieldType::OLAP_FIELD_TYPE_DECIMAL128I:
         ptr = doris::vectorized::PredicateColumnType<TYPE_DECIMAL128I>::create();
         break;
+    case FieldType::OLAP_FIELD_TYPE_DECIMAL256:
+        ptr = doris::vectorized::PredicateColumnType<TYPE_DECIMAL256>::create();
+        break;
     case FieldType::OLAP_FIELD_TYPE_ARRAY:
         ptr = doris::vectorized::ColumnArray::create(
                 get_predicate_column_ptr(*field.get_sub_field(0), reader_type),
diff --git a/be/src/olap/tablet_schema.cpp b/be/src/olap/tablet_schema.cpp
index 6f25b20e650eff..54c984e979850d 100644
--- a/be/src/olap/tablet_schema.cpp
+++ b/be/src/olap/tablet_schema.cpp
@@ -92,6 +92,8 @@ FieldType TabletColumn::get_field_type_by_string(const std::string& type_str) {
         type = FieldType::OLAP_FIELD_TYPE_DECIMAL64;
     } else if (0 == upper_type_str.compare("DECIMAL128I")) {
         type = FieldType::OLAP_FIELD_TYPE_DECIMAL128I;
+    } else if (0 == upper_type_str.compare("DECIMAL256")) {
+        type = FieldType::OLAP_FIELD_TYPE_DECIMAL256;
     } else if (0 == upper_type_str.compare(0, 7, "DECIMAL")) {
         type = FieldType::OLAP_FIELD_TYPE_DECIMAL;
     } else if (0 == upper_type_str.compare(0, 7, "VARCHAR")) {
@@ -226,6 +228,9 @@ std::string TabletColumn::get_string_by_field_type(FieldType type) {
     case FieldType::OLAP_FIELD_TYPE_DECIMAL128I:
         return "DECIMAL128I";
 
+    case FieldType::OLAP_FIELD_TYPE_DECIMAL256:
+        return "DECIMAL256";
+
     case FieldType::OLAP_FIELD_TYPE_VARCHAR:
         return "VARCHAR";
 
diff --git a/be/src/olap/types.cpp b/be/src/olap/types.cpp
index 2c92bd3f2c81b5..b095d830e7079f 100644
--- a/be/src/olap/types.cpp
+++ b/be/src/olap/types.cpp
@@ -98,6 +98,7 @@ const TypeInfo* get_scalar_type_info(FieldType field_type) {
             get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_JSONB>(),
             get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_VARIANT>(),
             get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_AGG_STATE>(),
+            get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_DECIMAL256>(),
             nullptr};
     return field_type_array[int(field_type)];
 }
diff --git a/be/src/olap/types.h b/be/src/olap/types.h
index bb54959aee8b3f..2fec9fc035405c 100644
--- a/be/src/olap/types.h
+++ b/be/src/olap/types.h
@@ -53,6 +53,7 @@
 #include "util/string_parser.hpp"
 #include "util/types.h"
 #include "vec/common/arena.h"
+#include "vec/core/wide_integer.h"
 #include "vec/runtime/vdatetime_value.h"
 
 namespace doris {
@@ -690,6 +691,11 @@ struct CppTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL128I> {
     using UnsignedCppType = uint128_t;
 };
 template <>
+struct CppTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL256> {
+    using CppType = Int256;
+    using UnsignedCppType = wide::UInt256;
+};
+template <>
 struct CppTypeTraits<FieldType::OLAP_FIELD_TYPE_DATE> {
     using CppType = uint24_t;
     using UnsignedCppType = uint24_t;
@@ -1083,6 +1089,30 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL128I>
     }
 };
 
+template <>
+struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL256>
+        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL256> {
+    static Status from_string(void* buf, const std::string& scan_key, const int precision,
+                              const int scale) {
+        StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
+        auto value = StringParser::string_to_decimal<TYPE_DECIMAL256>(
+                scan_key.c_str(), scan_key.size(), 76, scale, &result);
+        if (result == StringParser::PARSE_FAILURE) {
+            return Status::Error<ErrorCode::INVALID_ARGUMENT>(
+                    "FieldTypeTraits<OLAP_FIELD_TYPE_DECIMAL128I>::from_string meet PARSE_FAILURE");
+        }
+        *reinterpret_cast<Int256*>(buf) = value;
+        return Status::OK();
+    }
+    static std::string to_string(const void* src) {
+        return "";
+        // auto value = reinterpret_cast<const wide::Int256*>(src);
+        // fmt::memory_buffer buffer;
+        // fmt::format_to(buffer, "{}", *value);
+        // return std::string(buffer.data(), buffer.size());
+    }
+};
+
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATE>
         : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DATE> {
diff --git a/be/src/olap/utils.h b/be/src/olap/utils.h
index 75df35e32ac860..f8e2e1fbe9936e 100644
--- a/be/src/olap/utils.h
+++ b/be/src/olap/utils.h
@@ -257,6 +257,7 @@ constexpr bool is_numeric_type(const FieldType& field_type) {
            field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL32 ||
            field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL64 ||
            field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL128I ||
+           field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL256 ||
            field_type == FieldType::OLAP_FIELD_TYPE_BOOL;
 }
 
diff --git a/be/src/runtime/primitive_type.cpp b/be/src/runtime/primitive_type.cpp
index 82a189107d9301..304919dcdc58fd 100644
--- a/be/src/runtime/primitive_type.cpp
+++ b/be/src/runtime/primitive_type.cpp
@@ -125,6 +125,9 @@ PrimitiveType thrift_to_type(TPrimitiveType::type ttype) {
     case TPrimitiveType::DECIMAL128I:
         return TYPE_DECIMAL128I;
 
+    case TPrimitiveType::DECIMAL256:
+        return TYPE_DECIMAL256;
+
     case TPrimitiveType::CHAR:
         return TYPE_CHAR;
 
diff --git a/be/src/runtime/types.cpp b/be/src/runtime/types.cpp
index 4cb3d3ef5bb630..6243189af00c36 100644
--- a/be/src/runtime/types.cpp
+++ b/be/src/runtime/types.cpp
@@ -46,7 +46,8 @@ TypeDescriptor::TypeDescriptor(const std::vector<TTypeNode>& types, int* idx)
             DCHECK(scalar_type.__isset.len);
             len = scalar_type.len;
         } else if (type == TYPE_DECIMALV2 || type == TYPE_DECIMAL32 || type == TYPE_DECIMAL64 ||
-                   type == TYPE_DECIMAL128I || type == TYPE_DATETIMEV2 || type == TYPE_TIMEV2) {
+                   type == TYPE_DECIMAL128I || type == TYPE_DECIMAL256 || type == TYPE_DATETIMEV2 ||
+                   type == TYPE_TIMEV2) {
             DCHECK(scalar_type.__isset.precision);
             DCHECK(scalar_type.__isset.scale);
             precision = scalar_type.precision;
diff --git a/be/src/vec/core/field.h b/be/src/vec/core/field.h
index b997d9ddc22245..49b8e8d7a3ec53 100644
--- a/be/src/vec/core/field.h
+++ b/be/src/vec/core/field.h
@@ -319,7 +319,8 @@ class Field {
             Bitmap = 27,
             HyperLogLog = 28,
             QuantileState = 29,
-            Decimal256 = 30,
+            Int256 = 30,
+            Decimal256 = 31,
         };
 
         static const int MIN_NON_POD = 16;
@@ -643,6 +644,9 @@ class Field {
         case Types::Decimal128I:
             f(field.template get<DecimalField<Decimal128I>>());
             return;
+        case Types::Decimal256:
+            f(field.template get<DecimalField<Decimal256>>());
+            return;
         case Types::VariantMap:
             f(field.template get<VariantMap>());
             return;
@@ -780,6 +784,10 @@ struct Field::TypeToEnum<Int128> {
     static constexpr Types::Which value = Types::Int128;
 };
 template <>
+struct Field::TypeToEnum<Int256> {
+    static constexpr Types::Which value = Types::Int256;
+};
+template <>
 struct Field::TypeToEnum<Float64> {
     static constexpr Types::Which value = Types::Float64;
 };
diff --git a/be/src/vec/core/types.h b/be/src/vec/core/types.h
index 1d7a597f6d493b..1c59c4c9cb8339 100644
--- a/be/src/vec/core/types.h
+++ b/be/src/vec/core/types.h
@@ -96,7 +96,8 @@ enum class TypeIndex {
     QuantileState = 42,
     Time = 43,
     AggState = 44,
-    Decimal256
+    Decimal256 = 45,
+    Int256
 };
 
 struct Consted {
@@ -281,10 +282,21 @@ struct TypeName<Int128> {
     static const char* get() { return "Int128"; }
 };
 template <>
+inline constexpr bool IsNumber<Int256> = true;
+template <>
+struct TypeName<Int256> {
+    static const char* get() { return "Int256"; }
+};
+template <>
 struct TypeId<Int128> {
     static constexpr const TypeIndex value = TypeIndex::Int128;
 };
 
+template <>
+struct TypeId<Int256> {
+    static constexpr const TypeIndex value = TypeIndex::Int256;
+};
+
 using Date = Int64;
 using DateTime = Int64;
 using DateV2 = UInt32;
@@ -634,8 +646,11 @@ struct Decimal<Int256> {
         constexpr auto precision =
                 std::is_same_v<T, Int32>
                         ? BeConsts::MAX_DECIMAL32_PRECISION
-                        : (std::is_same_v<T, Int64> ? BeConsts::MAX_DECIMAL64_PRECISION
-                                                    : BeConsts::MAX_DECIMAL128_PRECISION);
+                        : (std::is_same_v<T, Int64>
+                                   ? BeConsts::MAX_DECIMAL64_PRECISION
+                                   : (std::is_same_v<T, Int128>
+                                              ? BeConsts::MAX_DECIMAL128_PRECISION
+                                              : BeConsts::MAX_DECIMAL256_PRECISION));
         return precision + 1 // Add a space for decimal place
                + 1           // Add a space for leading 0
                + 1;          // Add a space for negative sign
@@ -895,6 +910,11 @@ constexpr bool IsDecimal128I = false;
 template <>
 inline constexpr bool IsDecimal128I<Decimal128I> = true;
 
+template <typename T>
+constexpr bool IsDecimal256 = false;
+template <>
+inline constexpr bool IsDecimal256<Decimal256> = true;
+
 template <typename T>
 constexpr bool IsDecimalV2 = IsDecimal128<T> && !IsDecimal128I<T>;
 
@@ -959,6 +979,8 @@ inline const char* getTypeName(TypeIndex idx) {
         return TypeName<Int64>::get();
     case TypeIndex::Int128:
         return TypeName<Int128>::get();
+    case TypeIndex::Int256:
+        return TypeName<Int256>::get();
     case TypeIndex::Float32:
         return TypeName<Float32>::get();
     case TypeIndex::Float64:
diff --git a/be/src/vec/core/wide_integer.h b/be/src/vec/core/wide_integer.h
index c13e0092d47c72..7203a3db9886b3 100644
--- a/be/src/vec/core/wide_integer.h
+++ b/be/src/vec/core/wide_integer.h
@@ -129,6 +129,7 @@ class integer {
 };
 
 using Int256 = integer<256, signed>;
+using UInt256 = integer<256, unsigned>;
 
 template <typename T>
 static constexpr bool ArithmeticConcept() noexcept;
diff --git a/be/src/vec/data_types/data_type_decimal.cpp b/be/src/vec/data_types/data_type_decimal.cpp
index f5751a7f1f0161..a0ebbf15eb2812 100644
--- a/be/src/vec/data_types/data_type_decimal.cpp
+++ b/be/src/vec/data_types/data_type_decimal.cpp
@@ -168,10 +168,10 @@ bool DataTypeDecimal<T>::parse_from_string(const std::string& str, T* res) const
 
 DataTypePtr create_decimal(UInt64 precision_value, UInt64 scale_value, bool use_v2) {
     if (precision_value < min_decimal_precision() ||
-        precision_value > max_decimal_precision<Decimal128>()) {
+        precision_value > max_decimal_precision<Decimal256>()) {
         throw doris::Exception(doris::ErrorCode::NOT_IMPLEMENTED_ERROR,
                                "Wrong precision {}, min: {}, max: {}", precision_value,
-                               min_decimal_precision(), max_decimal_precision<Decimal128>());
+                               min_decimal_precision(), max_decimal_precision<Decimal256>());
     }
 
     if (static_cast<UInt64>(scale_value) > precision_value) {
diff --git a/be/src/vec/data_types/data_type_factory.cpp b/be/src/vec/data_types/data_type_factory.cpp
index 479b1261de72bf..e2c127d964389e 100644
--- a/be/src/vec/data_types/data_type_factory.cpp
+++ b/be/src/vec/data_types/data_type_factory.cpp
@@ -187,6 +187,7 @@ DataTypePtr DataTypeFactory::create_data_type(const TypeDescriptor& col_desc, bo
     case TYPE_DECIMAL32:
     case TYPE_DECIMAL64:
     case TYPE_DECIMAL128I:
+    case TYPE_DECIMAL256:
         nested = vectorized::create_decimal(col_desc.precision, col_desc.scale, false);
         break;
     // Just Mock A NULL Type in Vec Exec Engine
@@ -398,6 +399,7 @@ DataTypePtr DataTypeFactory::_create_primitive_data_type(const FieldType& type,
     case FieldType::OLAP_FIELD_TYPE_DECIMAL32:
     case FieldType::OLAP_FIELD_TYPE_DECIMAL64:
     case FieldType::OLAP_FIELD_TYPE_DECIMAL128I:
+    case FieldType::OLAP_FIELD_TYPE_DECIMAL256:
         result = vectorized::create_decimal(precision, scale, false);
         break;
     default:
@@ -483,10 +485,10 @@ DataTypePtr DataTypeFactory::create_data_type(const PColumnMeta& pcolumn) {
         nested = std::make_shared<DataTypeDecimal<Decimal128I>>(pcolumn.decimal_param().precision(),
                                                                 pcolumn.decimal_param().scale());
         break;
-    // case PGenericType::DECIMAL256:
-    //     nested = std::make_shared<DataTypeDecimal<Decimal256>>(pcolumn.decimal_param().precision(),
-    //                                                             pcolumn.decimal_param().scale());
-    //     break;
+    case PGenericType::DECIMAL256:
+        nested = std::make_shared<DataTypeDecimal<Decimal256>>(pcolumn.decimal_param().precision(),
+                                                               pcolumn.decimal_param().scale());
+        break;
     case PGenericType::BITMAP:
         nested = std::make_shared<DataTypeBitMap>();
         break;
diff --git a/be/src/vec/olap/olap_data_convertor.cpp b/be/src/vec/olap/olap_data_convertor.cpp
index 99304e69376627..0870f89d671139 100644
--- a/be/src/vec/olap/olap_data_convertor.cpp
+++ b/be/src/vec/olap/olap_data_convertor.cpp
@@ -131,9 +131,9 @@ OlapBlockDataConvertor::create_olap_column_data_convertor(const TabletColumn& co
     case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
         return std::make_unique<OlapColumnDataConvertorDecimalV3<Decimal128I>>();
     }
-    // case FieldType::OLAP_FIELD_TYPE_DECIMAL256: {
-    //     return std::make_unique<OlapColumnDataConvertorDecimalV3<Decimal256>>();
-    // }
+    case FieldType::OLAP_FIELD_TYPE_DECIMAL256: {
+        return std::make_unique<OlapColumnDataConvertorDecimalV3<Decimal256>>();
+    }
     case FieldType::OLAP_FIELD_TYPE_JSONB: {
         return std::make_unique<OlapColumnDataConvertorVarChar>(true);
     }
diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java
index 1cf7e887d28cff..ad24117e878019 100644
--- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java
+++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ColumnType.java
@@ -54,6 +54,7 @@ public enum Type {
         DECIMAL32(4),
         DECIMAL64(8),
         DECIMAL128(16),
+        DECIMAL256(32),
         STRING(-1),
         ARRAY(-1),
         MAP(-1),
@@ -309,13 +310,17 @@ public static ColumnType parseType(String columnName, String hiveType) {
                             type = Type.DECIMAL64;
                         } else if (lowerCaseType.startsWith("decimal128")) {
                             type = Type.DECIMAL128;
+                        } else if (lowerCaseType.startsWith("decimal256")) {
+                            type = Type.DECIMAL256;
                         } else {
                             if (precision <= MAX_DECIMAL32_PRECISION) {
                                 type = Type.DECIMAL32;
                             } else if (precision <= MAX_DECIMAL64_PRECISION) {
                                 type = Type.DECIMAL64;
-                            } else {
+                            } else if (precision <= MAX_DECIMAL128_PRECISION) {
                                 type = Type.DECIMAL128;
+                            } else {
+                                type = Type.DECIMAL256;
                             }
                         }
                     }
diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ScanPredicate.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ScanPredicate.java
index e82f05c7d0a367..92fd571d272337 100644
--- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ScanPredicate.java
+++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/ScanPredicate.java
@@ -106,6 +106,7 @@ private Object inspectObject() {
                 case DECIMAL32:
                 case DECIMAL64:
                 case DECIMAL128:
+                case DECIMAL256:
                     return TypeNativeBytes.getDecimal(Arrays.copyOf(valueBytes, valueBytes.length), scale);
                 case CHAR:
                 case VARCHAR:
diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/VectorColumn.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/VectorColumn.java
index 3998a1a3270aff..74539a1a0c52a3 100644
--- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/VectorColumn.java
+++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/vec/VectorColumn.java
@@ -282,6 +282,7 @@ public int appendNull(ColumnType.Type typeValue) {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 return appendDecimal(new BigDecimal(0));
             case DATEV2:
                 return appendDate(LocalDate.MIN);
@@ -619,6 +620,7 @@ public void appendValue(ColumnValue o) {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 appendDecimal(o.getDecimal());
                 break;
             case DATEV2:
@@ -681,6 +683,7 @@ public void dump(StringBuilder sb, int i) {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 sb.append(getDecimal(i));
                 break;
             case DATEV2:
diff --git a/fe/fe-common/src/main/java/org/apache/doris/catalog/PrimitiveType.java b/fe/fe-common/src/main/java/org/apache/doris/catalog/PrimitiveType.java
index 5f34bb65c96341..a5d6145a8fc48f 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/catalog/PrimitiveType.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/catalog/PrimitiveType.java
@@ -55,6 +55,7 @@ public enum PrimitiveType {
     DECIMAL32("DECIMAL32", 4, TPrimitiveType.DECIMAL32, true),
     DECIMAL64("DECIMAL64", 8, TPrimitiveType.DECIMAL64, true),
     DECIMAL128("DECIMAL128", 16, TPrimitiveType.DECIMAL128I, true),
+    DECIMAL256("DECIMAL256", 32, TPrimitiveType.DECIMAL256, true),
     TIME("TIME", 8, TPrimitiveType.TIME, false),
     // these following types are stored as object binary in BE.
     HLL("HLL", 16, TPrimitiveType.HLL, true),
@@ -95,6 +96,7 @@ public enum PrimitiveType {
         builder.add(DECIMAL32);
         builder.add(DECIMAL64);
         builder.add(DECIMAL128);
+        builder.add(DECIMAL256);
         builder.add(DATETIMEV2);
         typeWithPrecision = builder.build();
     }
@@ -124,6 +126,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(NULL_TYPE, DECIMAL32);
         builder.put(NULL_TYPE, DECIMAL64);
         builder.put(NULL_TYPE, DECIMAL128);
+        builder.put(NULL_TYPE, DECIMAL256);
         builder.put(NULL_TYPE, CHAR);
         builder.put(NULL_TYPE, VARCHAR);
         builder.put(NULL_TYPE, STRING);
@@ -149,6 +152,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(BOOLEAN, DECIMAL32);
         builder.put(BOOLEAN, DECIMAL64);
         builder.put(BOOLEAN, DECIMAL128);
+        builder.put(BOOLEAN, DECIMAL256);
         builder.put(BOOLEAN, VARCHAR);
         builder.put(BOOLEAN, STRING);
         // Tinyint
@@ -168,6 +172,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(TINYINT, DECIMAL32);
         builder.put(TINYINT, DECIMAL64);
         builder.put(TINYINT, DECIMAL128);
+        builder.put(TINYINT, DECIMAL256);
         builder.put(TINYINT, VARCHAR);
         builder.put(TINYINT, STRING);
         builder.put(TINYINT, TIME);
@@ -189,6 +194,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(SMALLINT, DECIMAL32);
         builder.put(SMALLINT, DECIMAL64);
         builder.put(SMALLINT, DECIMAL128);
+        builder.put(SMALLINT, DECIMAL256);
         builder.put(SMALLINT, VARCHAR);
         builder.put(SMALLINT, STRING);
         builder.put(SMALLINT, TIME);
@@ -210,6 +216,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(INT, DECIMAL32);
         builder.put(INT, DECIMAL64);
         builder.put(INT, DECIMAL128);
+        builder.put(INT, DECIMAL256);
         builder.put(INT, VARCHAR);
         builder.put(INT, STRING);
         builder.put(INT, TIME);
@@ -231,6 +238,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(BIGINT, DECIMAL32);
         builder.put(BIGINT, DECIMAL64);
         builder.put(BIGINT, DECIMAL128);
+        builder.put(BIGINT, DECIMAL256);
         builder.put(BIGINT, VARCHAR);
         builder.put(BIGINT, STRING);
         builder.put(BIGINT, TIME);
@@ -252,6 +260,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(LARGEINT, DECIMAL32);
         builder.put(LARGEINT, DECIMAL64);
         builder.put(LARGEINT, DECIMAL128);
+        builder.put(LARGEINT, DECIMAL256);
         builder.put(LARGEINT, VARCHAR);
         builder.put(LARGEINT, STRING);
         builder.put(LARGEINT, TIME);
@@ -273,6 +282,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(FLOAT, DECIMAL32);
         builder.put(FLOAT, DECIMAL64);
         builder.put(FLOAT, DECIMAL128);
+        builder.put(FLOAT, DECIMAL256);
         builder.put(FLOAT, VARCHAR);
         builder.put(FLOAT, STRING);
         builder.put(FLOAT, TIME);
@@ -294,6 +304,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DOUBLE, DECIMAL32);
         builder.put(DOUBLE, DECIMAL64);
         builder.put(DOUBLE, DECIMAL128);
+        builder.put(DOUBLE, DECIMAL256);
         builder.put(DOUBLE, VARCHAR);
         builder.put(DOUBLE, STRING);
         builder.put(DOUBLE, TIME);
@@ -315,6 +326,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DATE, DECIMAL32);
         builder.put(DATE, DECIMAL64);
         builder.put(DATE, DECIMAL128);
+        builder.put(DATE, DECIMAL256);
         builder.put(DATE, VARCHAR);
         builder.put(DATE, STRING);
         // Datetime
@@ -334,6 +346,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DATETIME, DECIMAL32);
         builder.put(DATETIME, DECIMAL64);
         builder.put(DATETIME, DECIMAL128);
+        builder.put(DATETIME, DECIMAL256);
         builder.put(DATETIME, VARCHAR);
         builder.put(DATETIME, STRING);
         // DateV2
@@ -353,6 +366,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DATEV2, DECIMAL32);
         builder.put(DATEV2, DECIMAL64);
         builder.put(DATEV2, DECIMAL128);
+        builder.put(DATEV2, DECIMAL256);
         builder.put(DATEV2, VARCHAR);
         builder.put(DATEV2, STRING);
         // DatetimeV2
@@ -372,6 +386,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DATETIMEV2, DECIMAL32);
         builder.put(DATETIMEV2, DECIMAL64);
         builder.put(DATETIMEV2, DECIMAL128);
+        builder.put(DATETIMEV2, DECIMAL256);
         builder.put(DATETIMEV2, VARCHAR);
         builder.put(DATETIMEV2, STRING);
         // Char
@@ -392,6 +407,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(CHAR, DECIMAL32);
         builder.put(CHAR, DECIMAL64);
         builder.put(CHAR, DECIMAL128);
+        builder.put(CHAR, DECIMAL256);
         builder.put(CHAR, VARCHAR);
         builder.put(CHAR, STRING);
         builder.put(CHAR, TIME);
@@ -413,6 +429,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(VARCHAR, DECIMAL32);
         builder.put(VARCHAR, DECIMAL64);
         builder.put(VARCHAR, DECIMAL128);
+        builder.put(VARCHAR, DECIMAL256);
         builder.put(VARCHAR, VARCHAR);
         builder.put(VARCHAR, JSONB);
         builder.put(VARCHAR, VARIANT);
@@ -437,6 +454,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(STRING, DECIMAL32);
         builder.put(STRING, DECIMAL64);
         builder.put(STRING, DECIMAL128);
+        builder.put(STRING, DECIMAL256);
         builder.put(STRING, VARCHAR);
         builder.put(STRING, JSONB);
         builder.put(STRING, VARIANT);
@@ -457,6 +475,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DECIMALV2, DECIMAL32);
         builder.put(DECIMALV2, DECIMAL64);
         builder.put(DECIMALV2, DECIMAL128);
+        builder.put(DECIMALV2, DECIMAL256);
         builder.put(DECIMALV2, VARCHAR);
         builder.put(DECIMALV2, STRING);
 
@@ -472,6 +491,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DECIMAL32, DECIMAL32);
         builder.put(DECIMAL32, DECIMAL64);
         builder.put(DECIMAL32, DECIMAL128);
+        builder.put(DECIMAL32, DECIMAL256);
         builder.put(DECIMAL32, VARCHAR);
         builder.put(DECIMAL32, STRING);
 
@@ -487,6 +507,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DECIMAL64, DECIMAL32);
         builder.put(DECIMAL64, DECIMAL64);
         builder.put(DECIMAL64, DECIMAL128);
+        builder.put(DECIMAL64, DECIMAL256);
         builder.put(DECIMAL64, VARCHAR);
         builder.put(DECIMAL64, STRING);
 
@@ -502,6 +523,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(DECIMAL128, DECIMAL32);
         builder.put(DECIMAL128, DECIMAL64);
         builder.put(DECIMAL128, DECIMAL128);
+        builder.put(DECIMAL128, DECIMAL256);
         builder.put(DECIMAL128, VARCHAR);
         builder.put(DECIMAL128, STRING);
 
@@ -518,6 +540,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         builder.put(JSONB, DECIMAL32);
         builder.put(JSONB, DECIMAL64);
         builder.put(JSONB, DECIMAL128);
+        builder.put(JSONB, DECIMAL256);
         builder.put(JSONB, VARCHAR);
         builder.put(JSONB, STRING);
         builder.put(JSONB, VARIANT);
@@ -576,6 +599,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         numericTypes.add(DECIMAL32);
         numericTypes.add(DECIMAL64);
         numericTypes.add(DECIMAL128);
+        numericTypes.add(DECIMAL256);
 
         supportedTypes = Lists.newArrayList();
         supportedTypes.add(NULL_TYPE);
@@ -603,6 +627,7 @@ public static ImmutableSetMultimap<PrimitiveType, PrimitiveType> getImplicitCast
         supportedTypes.add(DECIMAL32);
         supportedTypes.add(DECIMAL64);
         supportedTypes.add(DECIMAL128);
+        supportedTypes.add(DECIMAL256);
         supportedTypes.add(BITMAP);
         supportedTypes.add(ARRAY);
         supportedTypes.add(MAP);
@@ -1055,6 +1080,8 @@ public static PrimitiveType fromThrift(TPrimitiveType tPrimitiveType) {
                 return DECIMAL64;
             case DECIMAL128I:
                 return DECIMAL128;
+            case DECIMAL256:
+                return DECIMAL256;
             case TIME:
                 return TIME;
             case TIMEV2:
@@ -1162,7 +1189,7 @@ public boolean isDecimalV2Type() {
     }
 
     public boolean isDecimalV3Type() {
-        return this == DECIMAL32 || this == DECIMAL64 || this == DECIMAL128;
+        return this == DECIMAL32 || this == DECIMAL64 || this == DECIMAL128 || this == DECIMAL256;
     }
 
     public boolean isNumericType() {
@@ -1271,6 +1298,7 @@ public MysqlColType toMysqlType() {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 return MysqlColType.MYSQL_TYPE_NEWDECIMAL;
             case STRING:
                 return MysqlColType.MYSQL_TYPE_BLOB;
@@ -1308,6 +1336,8 @@ public int getOlapColumnIndexSize() {
                 return 8;
             case DECIMAL128:
                 return 16;
+            case DECIMAL256:
+                return 32;
             default:
                 return this.getSlotSize();
         }
diff --git a/fe/fe-common/src/main/java/org/apache/doris/catalog/ScalarType.java b/fe/fe-common/src/main/java/org/apache/doris/catalog/ScalarType.java
index b8dc17d54639dc..d24950658a76f0 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/catalog/ScalarType.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/catalog/ScalarType.java
@@ -78,6 +78,7 @@ public class ScalarType extends Type {
     public static final int MAX_DECIMAL32_PRECISION = 9;
     public static final int MAX_DECIMAL64_PRECISION = 18;
     public static final int MAX_DECIMAL128_PRECISION = 38;
+    public static final int MAX_DECIMAL256_PRECISION = 76;
     public static final int DEFAULT_MIN_AVG_DECIMAL128_SCALE = 4;
     public static final int MAX_DATETIMEV2_SCALE = 6;
 
@@ -138,6 +139,7 @@ public static ScalarType createType(PrimitiveType type, int len, int precision,
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 return createDecimalV3Type(precision, scale);
             case DECIMALV2:
                 return createDecimalType(precision, scale);
@@ -210,6 +212,8 @@ public static ScalarType createType(PrimitiveType type) {
                 return DEFAULT_DECIMAL64;
             case DECIMAL128:
                 return DEFAULT_DECIMAL128;
+            case DECIMAL256:
+                return DEFAULT_DECIMAL256;
             case DECIMALV2:
                 return DEFAULT_DECIMALV2;
             case LARGEINT:
@@ -385,8 +389,10 @@ public static PrimitiveType getSuitableDecimalType(int precision, boolean decima
             return PrimitiveType.DECIMAL32;
         } else if (precision <= MAX_DECIMAL64_PRECISION) {
             return PrimitiveType.DECIMAL64;
-        } else {
+        } else if (precision <= MAX_DECIMAL128_PRECISION) {
             return PrimitiveType.DECIMAL128;
+        } else {
+            return PrimitiveType.DECIMAL256;
         }
     }
 
@@ -611,6 +617,7 @@ public String toSql(int depth) {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 String typeName = "decimalv3";
                 if (Strings.isNullOrEmpty(precisionStr)) {
                     stringBuilder.append(typeName).append("(").append(precision)
@@ -701,6 +708,7 @@ public void toThrift(TTypeDesc container) {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
             case DATETIMEV2: {
                 Preconditions.checkArgument(precision >= scale,
                         String.format("given precision %d is out of scale bound %d", precision, scale));
@@ -724,14 +732,14 @@ public void toThrift(TTypeDesc container) {
     public int decimalPrecision() {
         Preconditions.checkState(type == PrimitiveType.DECIMALV2 || type == PrimitiveType.DATETIMEV2
                 || type == PrimitiveType.TIMEV2 || type == PrimitiveType.DECIMAL32
-                || type == PrimitiveType.DECIMAL64 || type == PrimitiveType.DECIMAL128);
+                || type == PrimitiveType.DECIMAL64 || type == PrimitiveType.DECIMAL128 || type == PrimitiveType.DECIMAL256);
         return precision;
     }
 
     public int decimalScale() {
         Preconditions.checkState(type == PrimitiveType.DECIMALV2 || type == PrimitiveType.DATETIMEV2
                 || type == PrimitiveType.TIMEV2 || type == PrimitiveType.DECIMAL32
-                || type == PrimitiveType.DECIMAL64 || type == PrimitiveType.DECIMAL128);
+                || type == PrimitiveType.DECIMAL64 || type == PrimitiveType.DECIMAL128 || type == PrimitiveType.DECIMAL256);
         return scale;
     }
 
@@ -939,6 +947,8 @@ public Type getMaxResolutionType() {
             return createDecimalTypeInternal(MAX_DECIMAL64_PRECISION, scale, false);
         } else if (getPrimitiveType() == PrimitiveType.DECIMAL128) {
             return createDecimalTypeInternal(MAX_DECIMAL128_PRECISION, scale, false);
+        } else if (getPrimitiveType() == PrimitiveType.DECIMAL256) {
+            return createDecimalTypeInternal(MAX_DECIMAL256_PRECISION, scale, false);
         } else if (isLargeIntType()) {
             return ScalarType.LARGEINT;
         } else if (isDatetimeV2()) {
diff --git a/fe/fe-common/src/main/java/org/apache/doris/catalog/Type.java b/fe/fe-common/src/main/java/org/apache/doris/catalog/Type.java
index e75aec0d2a3c01..e667ae01ef27ae 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/catalog/Type.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/catalog/Type.java
@@ -87,6 +87,10 @@ public abstract class Type {
     public static final ScalarType DEFAULT_DECIMAL128 =
             ScalarType.createDecimalType(PrimitiveType.DECIMAL128, ScalarType.MAX_DECIMAL128_PRECISION,
                     ScalarType.DEFAULT_SCALE);
+
+    public static final ScalarType DEFAULT_DECIMAL256 =
+            ScalarType.createDecimalType(PrimitiveType.DECIMAL256, ScalarType.MAX_DECIMAL256_PRECISION,
+                    ScalarType.DEFAULT_SCALE);
     public static final ScalarType DEFAULT_DECIMALV3 = DEFAULT_DECIMAL32;
     public static final ScalarType DEFAULT_DATETIMEV2 = ScalarType.createDatetimeV2Type(0);
     public static final ScalarType DATETIMEV2 = DEFAULT_DATETIMEV2;
@@ -96,6 +100,7 @@ public abstract class Type {
     public static final ScalarType DECIMAL32 = DEFAULT_DECIMAL32;
     public static final ScalarType DECIMAL64 = DEFAULT_DECIMAL64;
     public static final ScalarType DECIMAL128 = DEFAULT_DECIMAL128;
+    public static final ScalarType DECIMAL256 = DEFAULT_DECIMAL256;
     public static final ScalarType JSONB = new ScalarType(PrimitiveType.JSONB);
     // (ScalarType) ScalarType.createDecimalTypeInternal(-1, -1);
     public static final ScalarType DEFAULT_VARCHAR = ScalarType.createVarcharType(-1);
@@ -149,6 +154,7 @@ public abstract class Type {
         numericTypes.add(DECIMAL32);
         numericTypes.add(DECIMAL64);
         numericTypes.add(DECIMAL128);
+        numericTypes.add(DECIMAL256);
 
         numericDateTimeTypes = Lists.newArrayList();
         numericDateTimeTypes.add(DATE);
@@ -198,6 +204,7 @@ public abstract class Type {
         arraySubTypes.add(DECIMAL32);
         arraySubTypes.add(DECIMAL64);
         arraySubTypes.add(DECIMAL128);
+        arraySubTypes.add(DECIMAL256);
 
         mapSubTypes = Lists.newArrayList();
         mapSubTypes.add(BOOLEAN);
@@ -208,6 +215,7 @@ public abstract class Type {
         mapSubTypes.add(DECIMAL32); // same DEFAULT_DECIMALV3
         mapSubTypes.add(DECIMAL64);
         mapSubTypes.add(DECIMAL128);
+        mapSubTypes.add(DECIMAL256);
         mapSubTypes.add(DATE);
         mapSubTypes.add(DATETIME);
         mapSubTypes.add(DATEV2);
@@ -226,6 +234,7 @@ public abstract class Type {
         structSubTypes.add(DECIMAL32); // same DEFAULT_DECIMALV3
         structSubTypes.add(DECIMAL64);
         structSubTypes.add(DECIMAL128);
+        structSubTypes.add(DECIMAL256);
         structSubTypes.add(DATE);
         structSubTypes.add(DATETIME);
         structSubTypes.add(DATEV2);
@@ -260,6 +269,7 @@ public abstract class Type {
                     .put(PrimitiveType.DECIMAL32, Sets.newHashSet(BigDecimal.class))
                     .put(PrimitiveType.DECIMAL64, Sets.newHashSet(BigDecimal.class))
                     .put(PrimitiveType.DECIMAL128, Sets.newHashSet(BigDecimal.class))
+                    .put(PrimitiveType.DECIMAL256, Sets.newHashSet(BigDecimal.class))
                     .put(PrimitiveType.ARRAY, Sets.newHashSet(ArrayList.class))
                     .put(PrimitiveType.MAP, Sets.newHashSet(HashMap.class))
                     .build();
@@ -391,7 +401,7 @@ public boolean isDecimalV3OrContainsDecimalV3() {
 
     public boolean isDecimalV3() {
         return isScalarType(PrimitiveType.DECIMAL32) || isScalarType(PrimitiveType.DECIMAL64)
-                || isScalarType(PrimitiveType.DECIMAL128);
+                || isScalarType(PrimitiveType.DECIMAL128) || isScalarType(PrimitiveType.DECIMAL256);
     }
 
     public boolean isDatetimeV2() {
@@ -788,6 +798,8 @@ public static Type getNextNumType(Type t) {
                 return DECIMAL64;
             case DECIMAL128:
                 return DECIMAL128;
+            case DECIMAL256:
+                return DECIMAL256;
             default:
                 return INVALID;
         }
@@ -880,6 +892,8 @@ public static Type fromPrimitiveType(PrimitiveType type) {
                 return Type.DECIMAL64;
             case DECIMAL128:
                 return Type.DECIMAL128;
+            case DECIMAL256:
+                return Type.DECIMAL256;
             case CHAR:
                 return Type.CHAR;
             case VARCHAR:
@@ -1125,6 +1139,7 @@ public Integer getPrecision() {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
             case DATETIMEV2:
             case TIMEV2:
                 return t.decimalPrecision();
@@ -1161,6 +1176,7 @@ public Integer getDecimalDigits() {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 return t.decimalScale();
             default:
                 return null;
@@ -1195,6 +1211,7 @@ public Integer getNumPrecRadix() {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 return 10;
             default:
                 // everything else (including boolean and string) is null
@@ -1262,6 +1279,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[BOOLEAN.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[BOOLEAN.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[BOOLEAN.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[BOOLEAN.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[BOOLEAN.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         // TINYINT
@@ -1283,6 +1301,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[TINYINT.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL32;
         compatibilityMatrix[TINYINT.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[TINYINT.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[TINYINT.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[TINYINT.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TINYINT.ordinal()][TIME.ordinal()] = PrimitiveType.DOUBLE;
         compatibilityMatrix[TINYINT.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1310,6 +1329,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[SMALLINT.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL32;
         compatibilityMatrix[SMALLINT.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[SMALLINT.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[SMALLINT.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[SMALLINT.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[SMALLINT.ordinal()][TIME.ordinal()] = PrimitiveType.DOUBLE;
         compatibilityMatrix[SMALLINT.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1340,6 +1360,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[INT.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL32;
         compatibilityMatrix[INT.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[INT.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[INT.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[INT.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[INT.ordinal()][TIME.ordinal()] = PrimitiveType.DOUBLE;
         compatibilityMatrix[INT.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1371,6 +1392,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[BIGINT.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[BIGINT.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[BIGINT.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[BIGINT.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[BIGINT.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[BIGINT.ordinal()][TIME.ordinal()] = PrimitiveType.DOUBLE;
         compatibilityMatrix[BIGINT.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1394,6 +1416,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[LARGEINT.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL128;
         compatibilityMatrix[LARGEINT.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL128;
         compatibilityMatrix[LARGEINT.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[LARGEINT.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[LARGEINT.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[LARGEINT.ordinal()][TIME.ordinal()] = PrimitiveType.DOUBLE;
         compatibilityMatrix[LARGEINT.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1416,6 +1439,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[FLOAT.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL128;
         compatibilityMatrix[FLOAT.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL128;
         compatibilityMatrix[FLOAT.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[FLOAT.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[FLOAT.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[FLOAT.ordinal()][TIME.ordinal()] = PrimitiveType.DOUBLE;
         compatibilityMatrix[FLOAT.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1434,6 +1458,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DOUBLE.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL128;
         compatibilityMatrix[DOUBLE.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL128;
         compatibilityMatrix[DOUBLE.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DOUBLE.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DOUBLE.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DOUBLE.ordinal()][TIME.ordinal()] = PrimitiveType.DOUBLE;
         compatibilityMatrix[DOUBLE.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1456,6 +1481,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DATE.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL32;
         compatibilityMatrix[DATE.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DATE.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DATE.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DATE.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATE.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATE.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1476,6 +1502,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DATEV2.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL32;
         compatibilityMatrix[DATEV2.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DATEV2.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DATEV2.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DATEV2.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATEV2.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATEV2.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1495,6 +1522,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DATETIME.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DATETIME.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DATETIME.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DATETIME.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DATETIME.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATETIME.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATETIME.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1514,6 +1542,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DATETIMEV2.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DATETIMEV2.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DATETIMEV2.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DATETIMEV2.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DATETIMEV2.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATETIMEV2.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DATETIMEV2.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1533,6 +1562,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[CHAR.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[CHAR.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[CHAR.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[CHAR.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[CHAR.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[CHAR.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[CHAR.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1548,6 +1578,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[VARCHAR.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARCHAR.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARCHAR.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[VARCHAR.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARCHAR.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARCHAR.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARCHAR.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1571,6 +1602,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[STRING.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[STRING.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[STRING.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[STRING.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[STRING.ordinal()][JSONB.ordinal()] = PrimitiveType.STRING;
         compatibilityMatrix[STRING.ordinal()][VARIANT.ordinal()] = PrimitiveType.STRING;
         compatibilityMatrix[STRING.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1580,6 +1612,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[JSONB.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[JSONB.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[JSONB.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[JSONB.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[JSONB.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[JSONB.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[JSONB.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1596,6 +1629,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[VARIANT.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARIANT.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARIANT.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[VARIANT.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARIANT.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARIANT.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[VARIANT.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1622,6 +1656,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DECIMALV2.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DECIMALV2.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DECIMALV2.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMALV2.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DECIMALV2.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         // DECIMAL32
@@ -1636,6 +1671,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DECIMAL32.ordinal()][DECIMALV2.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DECIMAL32.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DECIMAL32.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DECIMAL32.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DECIMAL32.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         // DECIMAL64
@@ -1650,6 +1686,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DECIMAL64.ordinal()][DECIMALV2.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DECIMAL64.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL64;
         compatibilityMatrix[DECIMAL64.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DECIMAL64.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DECIMAL64.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         // DECIMAL128
@@ -1664,8 +1701,24 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[DECIMAL128.ordinal()][DECIMALV2.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[DECIMAL128.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL128;
         compatibilityMatrix[DECIMAL128.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL128;
+        compatibilityMatrix[DECIMAL128.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.DECIMAL256;
         compatibilityMatrix[DECIMAL128.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
+        // DECIMAL256
+        compatibilityMatrix[DECIMAL256.ordinal()][HLL.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMAL256.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMAL256.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMAL256.ordinal()][DATEV2.ordinal()] = PrimitiveType.DECIMAL256;
+        compatibilityMatrix[DECIMAL256.ordinal()][DATETIMEV2.ordinal()] = PrimitiveType.DECIMAL256;
+        compatibilityMatrix[DECIMAL256.ordinal()][BITMAP.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMAL256.ordinal()][STRING.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMAL256.ordinal()][QUANTILE_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMAL256.ordinal()][DECIMALV2.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[DECIMAL256.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.DECIMAL256;
+        compatibilityMatrix[DECIMAL256.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.DECIMAL256;
+        compatibilityMatrix[DECIMAL256.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.DECIMAL256;
+        compatibilityMatrix[DECIMAL256.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
+
         // HLL
         compatibilityMatrix[HLL.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[HLL.ordinal()][TIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1679,6 +1732,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[HLL.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[HLL.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[HLL.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[HLL.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[HLL.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
 
@@ -1694,6 +1748,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[BITMAP.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[BITMAP.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[BITMAP.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[BITMAP.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[BITMAP.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         //QUANTILE_STATE
@@ -1705,6 +1760,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[QUANTILE_STATE.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[QUANTILE_STATE.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[QUANTILE_STATE.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[QUANTILE_STATE.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[QUANTILE_STATE.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         //AGG_STATE
@@ -1716,6 +1772,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[AGG_STATE.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[AGG_STATE.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[AGG_STATE.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[AGG_STATE.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         // TIME why here not???
         compatibilityMatrix[TIME.ordinal()][TIME.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1723,6 +1780,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[TIME.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIME.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIME.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[TIME.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIME.ordinal()][DATEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIME.ordinal()][DATETIMEV2.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIME.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
@@ -1732,6 +1790,7 @@ public Integer getNumPrecRadix() {
         compatibilityMatrix[TIMEV2.ordinal()][DECIMAL32.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIMEV2.ordinal()][DECIMAL64.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIMEV2.ordinal()][DECIMAL128.ordinal()] = PrimitiveType.INVALID_TYPE;
+        compatibilityMatrix[TIMEV2.ordinal()][DECIMAL256.ordinal()] = PrimitiveType.INVALID_TYPE;
         compatibilityMatrix[TIMEV2.ordinal()][AGG_STATE.ordinal()] = PrimitiveType.INVALID_TYPE;
 
         // Check all of the necessary entries that should be filled.
@@ -1793,6 +1852,8 @@ public Type getResultType() {
                 return DECIMAL64;
             case DECIMAL128:
                 return DECIMAL128;
+            case DECIMAL256:
+                return DECIMAL256;
             case STRING:
                 return STRING;
             case JSONB:
@@ -1870,8 +1931,10 @@ public static Type getCmpType(Type t1, Type t2) {
                 resultDecimalType = PrimitiveType.DECIMAL32;
             } else if (resultPrecision <= ScalarType.MAX_DECIMAL64_PRECISION) {
                 resultDecimalType = PrimitiveType.DECIMAL64;
-            } else {
+            } else if (resultPrecision <= ScalarType.MAX_DECIMAL128_PRECISION) {
                 resultDecimalType = PrimitiveType.DECIMAL128;
+            } else {
+                resultDecimalType = PrimitiveType.DECIMAL256;
             }
             if (resultPrecision <= ScalarType.MAX_DECIMAL128_PRECISION) {
                 return ScalarType.createDecimalType(resultDecimalType, resultPrecision, Math.max(
@@ -1976,6 +2039,8 @@ public Type getNumResultType() {
                 return Type.DECIMAL64;
             case DECIMAL128:
                 return Type.DECIMAL128;
+            case DECIMAL256:
+                return Type.DECIMAL256;
             default:
                 return Type.INVALID;
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CastExpr.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CastExpr.java
index 461647123aa12d..fff9a0e7fdfa86 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CastExpr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CastExpr.java
@@ -513,6 +513,7 @@ public CastExpr rewriteExpr(List<String> parameters, List<Expr> inputParamsExprs
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 // normal decimal
                 if (targetType.getPrecision() != 0) {
                     newTargetType = targetType;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ColumnDef.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ColumnDef.java
index 349be4e45ef614..0f5ff1e1313969 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/ColumnDef.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ColumnDef.java
@@ -500,6 +500,7 @@ public static void validateDefaultValue(Type type, String defaultValue, DefaultV
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 DecimalLiteral decimalLiteral = new DecimalLiteral(defaultValue);
                 decimalLiteral.checkPrecisionAndScale(scalarType.getScalarPrecision(), scalarType.getScalarScale());
                 break;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/Expr.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/Expr.java
index b4d550175fc1d5..382379c269577f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/Expr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/Expr.java
@@ -2502,6 +2502,8 @@ protected Type getActualScalarType(Type originType) {
             return Type.DECIMAL64;
         } else if (originType.getPrimitiveType() == PrimitiveType.DECIMAL128) {
             return Type.DECIMAL128;
+        } else if (originType.getPrimitiveType() == PrimitiveType.DECIMAL256) {
+            return Type.DECIMAL256;
         } else if (originType.getPrimitiveType() == PrimitiveType.DATETIMEV2) {
             return Type.DATETIMEV2;
         } else if (originType.getPrimitiveType() == PrimitiveType.DATEV2) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/LiteralExpr.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/LiteralExpr.java
index bde041aeefa73e..bfa50da17398e0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/LiteralExpr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/LiteralExpr.java
@@ -75,6 +75,7 @@ public static LiteralExpr create(String value, Type type) throws AnalysisExcepti
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 literalExpr = new DecimalLiteral(value);
                 break;
             case CHAR:
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/StringLiteral.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/StringLiteral.java
index 66747e0002f7fd..3119781c97cdb8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/StringLiteral.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/StringLiteral.java
@@ -245,6 +245,7 @@ protected Expr uncheckedCastTo(Type targetType) throws AnalysisException {
                 case DECIMAL32:
                 case DECIMAL64:
                 case DECIMAL128:
+                case DECIMAL256:
                     try {
                         DecimalLiteral res = new DecimalLiteral(new BigDecimal(value).stripTrailingZeros());
                         res.setType(targetType);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/TypeDef.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/TypeDef.java
index 9c80ee01ab1f45..ebc1f193f88828 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/TypeDef.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/TypeDef.java
@@ -292,6 +292,25 @@ private void analyzeScalarType(ScalarType scalarType)
                 }
                 break;
             }
+            case DECIMAL256: {
+                int precision = scalarType.decimalPrecision();
+                int scale = scalarType.decimalScale();
+                if (precision < 1 || precision > ScalarType.MAX_DECIMAL256_PRECISION) {
+                    throw new AnalysisException("Precision of decimal256 must between 1 and 76."
+                            + " Precision was set to: " + precision + ".");
+                }
+                // scale >= 0
+                if (scale < 0) {
+                    throw new AnalysisException("Scale of decimal must not be less than 0." + " Scale was set to: "
+                            + scale + ".");
+                }
+                // scale < precision
+                if (scale > precision) {
+                    throw new AnalysisException("Scale of decimal must be smaller than precision."
+                            + " Scale is " + scale + " and precision is " + precision);
+                }
+                break;
+            }
             case TIMEV2:
             case DATETIMEV2: {
                 int precision = scalarType.decimalPrecision();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateType.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateType.java
index 1d6b8628690cc1..499767e2fbd0dc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateType.java
@@ -54,6 +54,7 @@ public enum AggregateType {
         primitiveTypeList.add(PrimitiveType.DECIMAL32);
         primitiveTypeList.add(PrimitiveType.DECIMAL64);
         primitiveTypeList.add(PrimitiveType.DECIMAL128);
+        primitiveTypeList.add(PrimitiveType.DECIMAL256);
         compatibilityMap.put(SUM, EnumSet.copyOf(primitiveTypeList));
 
         primitiveTypeList.clear();
@@ -68,6 +69,7 @@ public enum AggregateType {
         primitiveTypeList.add(PrimitiveType.DECIMAL32);
         primitiveTypeList.add(PrimitiveType.DECIMAL64);
         primitiveTypeList.add(PrimitiveType.DECIMAL128);
+        primitiveTypeList.add(PrimitiveType.DECIMAL256);
         primitiveTypeList.add(PrimitiveType.DATE);
         primitiveTypeList.add(PrimitiveType.DATETIME);
         primitiveTypeList.add(PrimitiveType.DATEV2);
@@ -89,6 +91,7 @@ public enum AggregateType {
         primitiveTypeList.add(PrimitiveType.DECIMAL32);
         primitiveTypeList.add(PrimitiveType.DECIMAL64);
         primitiveTypeList.add(PrimitiveType.DECIMAL128);
+        primitiveTypeList.add(PrimitiveType.DECIMAL256);
         primitiveTypeList.add(PrimitiveType.DATE);
         primitiveTypeList.add(PrimitiveType.DATETIME);
         primitiveTypeList.add(PrimitiveType.DATEV2);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/AliasFunction.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/AliasFunction.java
index c2f6d466f220cf..882689dbfa8a12 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/AliasFunction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/AliasFunction.java
@@ -173,6 +173,7 @@ public void analyze() throws AnalysisException {
                     case DECIMAL32:
                     case DECIMAL64:
                     case DECIMAL128:
+                    case DECIMAL256:
                     case DECIMALV2:
                         if (!Strings.isNullOrEmpty(scalarType.getScalarPrecisionStr())) {
                             typeDefParams.add(scalarType.getScalarPrecisionStr());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
index 6de1b5a9d410d5..f8d0d3e716f0c6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
@@ -898,6 +898,7 @@ public String getSignatureString(Map<PrimitiveType, String> typeStringMap) {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 sb.append(String.format(typeStringMap.get(dataType), getPrecision(), getScale()));
                 break;
             case ARRAY:
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
index ae471ce8dd9019..c5d451bb21a17a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
@@ -88,6 +88,7 @@ public class Util {
         TYPE_STRING_MAP.put(PrimitiveType.DECIMAL32, "decimal(%d, %d)");
         TYPE_STRING_MAP.put(PrimitiveType.DECIMAL64, "decimal(%d, %d)");
         TYPE_STRING_MAP.put(PrimitiveType.DECIMAL128, "decimal(%d, %d)");
+        TYPE_STRING_MAP.put(PrimitiveType.DECIMAL256, "decimal(%d, %d)");
         TYPE_STRING_MAP.put(PrimitiveType.HLL, "varchar(%d)");
         TYPE_STRING_MAP.put(PrimitiveType.BOOLEAN, "bool");
         TYPE_STRING_MAP.put(PrimitiveType.BITMAP, "bitmap");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlSerializer.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlSerializer.java
index 59375ec4b67ed1..228f3891ce90e5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlSerializer.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlSerializer.java
@@ -265,7 +265,8 @@ private int getMysqlTypeLength(Type type) {
             case DECIMALV2:
             case DECIMAL32:
             case DECIMAL64:
-            case DECIMAL128: {
+            case DECIMAL128:
+            case DECIMAL256: {
                 // https://github.com/mysql/mysql-connector-j/blob/release/5.1/src/com/mysql/jdbc/ResultSetMetaData.java
                 // in function: int getPrecision(int column)
                 // f.getDecimals() > 0 ? clampedGetLength(f) - 1 + f.getPrecisionAdjustFactor()
@@ -296,6 +297,7 @@ public int getMysqlDecimals(Type type) {
             case DECIMAL32:
             case DECIMAL64:
             case DECIMAL128:
+            case DECIMAL256:
                 return ((ScalarType) type).decimalScale();
             case FLOAT:
             case DOUBLE:
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java
index 6ed045a3004f29..33fe8e85dd562e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java
@@ -209,7 +209,8 @@ private Map<String, Expression> evalOnBE(Map<String, Map<String, TExpr>> paramMa
                                     type = DateTimeV2Type.of(pScalarType.getScale());
                                 } else if (primitiveType == PrimitiveType.DECIMAL32
                                         || primitiveType == PrimitiveType.DECIMAL64
-                                        || primitiveType == PrimitiveType.DECIMAL128) {
+                                        || primitiveType == PrimitiveType.DECIMAL128
+                                        || primitiveType == PrimitiveType.DECIMAL256) {
                                     type = DecimalV3Type.createDecimalV3Type(
                                             pScalarType.getPrecision(), pScalarType.getScale());
                                 } else {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/cache/PartitionRange.java b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/PartitionRange.java
index 0c36cb69e768c9..0500814ed526c9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/cache/PartitionRange.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/cache/PartitionRange.java
@@ -189,6 +189,7 @@ public boolean init(Type type, LiteralExpr expr) {
                 case DECIMAL32:
                 case DECIMAL64:
                 case DECIMAL128:
+                case DECIMAL256:
                 case CHAR:
                 case VARCHAR:
                 case STRING: