diff --git a/be/src/exprs/bloom_filter_func.h b/be/src/exprs/bloom_filter_func.h
index a8330250ec084d5..bfcd7b0acf686fa 100644
--- a/be/src/exprs/bloom_filter_func.h
+++ b/be/src/exprs/bloom_filter_func.h
@@ -416,37 +416,10 @@ struct FixedStringFindOp : public StringFindOp {
     }
 };
 
-struct DateTimeFindOp : public CommonFindOp<VecDateTimeValue> {
-    bool find_olap_engine(const BloomFilterAdaptor& bloom_filter, const void* data) const {
-        VecDateTimeValue value;
-        value.from_olap_datetime(*reinterpret_cast<const uint64_t*>(data));
-        return bloom_filter.test(Slice((char*)&value, sizeof(VecDateTimeValue)));
-    }
-};
-
-// avoid violating C/C++ aliasing rules.
-// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=101684
-
-struct DateFindOp : public CommonFindOp<VecDateTimeValue> {
-    bool find_olap_engine(const BloomFilterAdaptor& bloom_filter, const void* data) const {
-        uint24_t date = *static_cast<const uint24_t*>(data);
-        uint64_t value = uint32_t(date);
-
-        VecDateTimeValue date_value;
-        date_value.from_olap_date(value);
-
-        return bloom_filter.test(Slice((char*)&date_value, sizeof(VecDateTimeValue)));
-    }
-};
-
 struct DecimalV2FindOp : public CommonFindOp<DecimalV2Value> {
     bool find_olap_engine(const BloomFilterAdaptor& bloom_filter, const void* data) const {
-        auto packed_decimal = *static_cast<const decimal12_t*>(data);
-        DecimalV2Value value;
-        int64_t int_value = packed_decimal.integer;
-        int32_t frac_value = packed_decimal.fraction;
-        value.from_olap_decimal(int_value, frac_value);
-
+        // Predicate column using decimalv2value as column value type, so not need convert here
+        DecimalV2Value value = *static_cast<const DecimalV2Value*>(data);
         constexpr int decimal_value_sz = sizeof(DecimalV2Value);
         char data_bytes[decimal_value_sz];
         memcpy(&data_bytes, &value, decimal_value_sz);
@@ -460,16 +433,6 @@ struct BloomFilterTypeTraits {
     using FindOp = CommonFindOp<T>;
 };
 
-template <>
-struct BloomFilterTypeTraits<TYPE_DATE> {
-    using FindOp = DateFindOp;
-};
-
-template <>
-struct BloomFilterTypeTraits<TYPE_DATETIME> {
-    using FindOp = DateTimeFindOp;
-};
-
 template <>
 struct BloomFilterTypeTraits<TYPE_DECIMALV2> {
     using FindOp = DecimalV2FindOp;
diff --git a/be/src/olap/column_predicate.h b/be/src/olap/column_predicate.h
index 05e84999a83102b..c0f83f32e97270e 100644
--- a/be/src/olap/column_predicate.h
+++ b/be/src/olap/column_predicate.h
@@ -54,6 +54,27 @@ enum class PredicateType {
     MATCH = 13,         // fulltext match
 };
 
+template <PrimitiveType primitive_type, typename ResultType>
+ResultType get_zone_map_value(void* data_ptr) {
+    ResultType res;
+    // DecimalV2's storage value is different from predicate or compute value type
+    // need convert it to DecimalV2Value
+    if constexpr (primitive_type == PrimitiveType::TYPE_DECIMALV2) {
+        decimal12_t decimal_12_t_value;
+        memcpy((char*)(&decimal_12_t_value), data_ptr, sizeof(decimal12_t));
+        res.from_olap_decimal(decimal_12_t_value.integer, decimal_12_t_value.fraction);
+    } else if constexpr (primitive_type == PrimitiveType::TYPE_DATE) {
+        static_assert(std::is_same_v<ResultType, VecDateTimeValue>);
+        res.from_olap_date(*reinterpret_cast<uint24_t*>(data_ptr));
+    } else if constexpr (primitive_type == PrimitiveType::TYPE_DATETIME) {
+        static_assert(std::is_same_v<ResultType, VecDateTimeValue>);
+        res.from_olap_datetime(*reinterpret_cast<uint64_t*>(data_ptr));
+    } else {
+        memcpy(&res, data_ptr, sizeof(ResultType));
+    }
+    return res;
+}
+
 inline std::string type_to_string(PredicateType type) {
     switch (type) {
     case PredicateType::UNKNOWN:
@@ -264,14 +285,6 @@ class ColumnPredicate {
     }
 
 protected:
-    // Just prevent access not align memory address coredump
-    template <class T>
-    T _get_zone_map_value(void* data_ptr) const {
-        T res;
-        memcpy(&res, data_ptr, sizeof(T));
-        return res;
-    }
-
     virtual std::string _debug_string() const = 0;
 
     uint32_t _column_id;
diff --git a/be/src/olap/comparison_predicate.h b/be/src/olap/comparison_predicate.h
index 522939370cf4443..d28ae299e1a1e57 100644
--- a/be/src/olap/comparison_predicate.h
+++ b/be/src/olap/comparison_predicate.h
@@ -32,7 +32,7 @@ namespace doris {
 template <PrimitiveType Type, PredicateType PT>
 class ComparisonPredicateBase : public ColumnPredicate {
 public:
-    using T = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using T = typename PrimitiveTypeTraits<Type>::CppType;
     ComparisonPredicateBase(uint32_t column_id, const T& value, bool opposite = false)
             : ColumnPredicate(column_id, opposite),
               _cached_code(_InvalidateCodeValue),
@@ -67,7 +67,20 @@ class ComparisonPredicateBase : public ColumnPredicate {
 
         roaring::Roaring roaring;
         bool exact_match = false;
-        Status status = iterator->seek_dictionary(&_value, &exact_match);
+
+        Status status;
+        if constexpr (Type == TYPE_DATE) {
+            uint24_t value = _value.to_olap_date();
+            status = iterator->seek_dictionary(&value, &exact_match);
+        } else if constexpr (Type == TYPE_DATETIME) {
+            auto value = _value.to_olap_datetime();
+            status = iterator->seek_dictionary(&value, &exact_match);
+        } else if constexpr (Type == TYPE_DECIMALV2) {
+            decimal12_t value {_value.int_value(), _value.frac_value()};
+            status = iterator->seek_dictionary(&value, &exact_match);
+        } else {
+            status = iterator->seek_dictionary(&_value, &exact_match);
+        }
         rowid_t seeked_ordinal = iterator->current_ordinal();
 
         return _bitmap_compare(status, exact_match, ordinal_limit, seeked_ordinal, iterator,
@@ -107,8 +120,22 @@ class ComparisonPredicateBase : public ColumnPredicate {
         }
 
         roaring::Roaring roaring;
-        RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, &_value, query_type,
-                                                           num_rows, &roaring));
+        if constexpr (Type == PrimitiveType::TYPE_DATE) {
+            uint24_t date = _value.to_olap_date();
+            RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, &date, query_type,
+                                                               num_rows, &roaring));
+        } else if constexpr (Type == PrimitiveType::TYPE_DATETIME) {
+            uint64_t date = _value.to_olap_datetime();
+            RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, &date, query_type,
+                                                               num_rows, &roaring));
+        } else if constexpr (Type == TYPE_DECIMALV2) {
+            decimal12_t value {_value.int_value(), _value.frac_value()};
+            RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, &value, query_type,
+                                                               num_rows, &roaring));
+        } else {
+            RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, &_value, query_type,
+                                                               num_rows, &roaring));
+        }
 
         // mask out null_bitmap, since NULL cmp VALUE will produce NULL
         //  and be treated as false in WHERE
@@ -150,17 +177,13 @@ class ComparisonPredicateBase : public ColumnPredicate {
         _evaluate_bit<true>(column, sel, size, flags);
     }
 
-    using WarpperFieldType = std::conditional_t<Type == TYPE_DATE, uint24_t, T>;
-
     bool evaluate_and(const std::pair<WrapperField*, WrapperField*>& statistic) const override {
         if (statistic.first->is_null()) {
             return true;
         }
 
-        T tmp_min_value {};
-        T tmp_max_value {};
-        memcpy((char*)(&tmp_min_value), statistic.first->cell_ptr(), sizeof(WarpperFieldType));
-        memcpy((char*)(&tmp_max_value), statistic.second->cell_ptr(), sizeof(WarpperFieldType));
+        T tmp_min_value = get_zone_map_value<Type, T>(statistic.first->cell_ptr());
+        T tmp_max_value = get_zone_map_value<Type, T>(statistic.second->cell_ptr());
 
         if constexpr (PT == PredicateType::EQ) {
             return _operator(tmp_min_value <= _value && tmp_max_value >= _value, true);
@@ -183,10 +206,8 @@ class ComparisonPredicateBase : public ColumnPredicate {
                 << " Type: " << Type << " sizeof(T): " << sizeof(T)
                 << " statistic.first->size(): " << statistic.first->size();
 
-        T tmp_min_value {};
-        T tmp_max_value {};
-        memcpy((char*)(&tmp_min_value), statistic.first->cell_ptr(), sizeof(WarpperFieldType));
-        memcpy((char*)(&tmp_max_value), statistic.second->cell_ptr(), sizeof(WarpperFieldType));
+        T tmp_min_value = get_zone_map_value<Type, T>(statistic.first->cell_ptr());
+        T tmp_max_value = get_zone_map_value<Type, T>(statistic.second->cell_ptr());
 
         if constexpr (PT == PredicateType::LT) {
             return _value > tmp_max_value;
@@ -206,10 +227,8 @@ class ComparisonPredicateBase : public ColumnPredicate {
             return false;
         }
 
-        T tmp_min_value {};
-        T tmp_max_value {};
-        memcpy((char*)(&tmp_min_value), statistic.first->cell_ptr(), sizeof(WarpperFieldType));
-        memcpy((char*)(&tmp_max_value), statistic.second->cell_ptr(), sizeof(WarpperFieldType));
+        T tmp_min_value = get_zone_map_value<Type, T>(statistic.first->cell_ptr());
+        T tmp_max_value = get_zone_map_value<Type, T>(statistic.second->cell_ptr());
 
         if constexpr (PT == PredicateType::EQ) {
             return tmp_min_value == _value && tmp_max_value == _value;
@@ -232,8 +251,16 @@ class ComparisonPredicateBase : public ColumnPredicate {
             if constexpr (std::is_same_v<T, StringRef>) {
                 return bf->test_bytes(_value.data, _value.size);
             } else {
-                return bf->test_bytes(const_cast<char*>(reinterpret_cast<const char*>(&_value)),
-                                      sizeof(WarpperFieldType));
+                // DecimalV2 using decimal12_t in bloom filter, should convert value to decimal12_t
+                if constexpr (Type == PrimitiveType::TYPE_DECIMALV2) {
+                    decimal12_t decimal12_t_val(_value.int_value(), _value.frac_value());
+                    return bf->test_bytes(
+                            const_cast<char*>(reinterpret_cast<const char*>(&decimal12_t_val)),
+                            sizeof(decimal12_t));
+                } else {
+                    return bf->test_bytes(const_cast<char*>(reinterpret_cast<const char*>(&_value)),
+                                          sizeof(T));
+                }
             }
         } else {
             LOG(FATAL) << "Bloom filter is not supported by predicate type.";
diff --git a/be/src/olap/decimal12.h b/be/src/olap/decimal12.h
index ad4a4e8c8760a3d..35d4a559e2b9c6e 100644
--- a/be/src/olap/decimal12.h
+++ b/be/src/olap/decimal12.h
@@ -81,6 +81,8 @@ struct decimal12_t {
         return std::string(buf);
     }
 
+    // Not modify this structure, ZoneMap use this from_string and to_string
+    // to serialize decimalv2 value to segment files
     Status from_string(const std::string& str) {
         integer = 0;
         fraction = 0;
diff --git a/be/src/olap/in_list_predicate.h b/be/src/olap/in_list_predicate.h
index cb733ab9b650bf8..dd67ff2c4a6122b 100644
--- a/be/src/olap/in_list_predicate.h
+++ b/be/src/olap/in_list_predicate.h
@@ -42,20 +42,7 @@ struct std::equal_to<doris::StringRef> {
         return lhs == rhs;
     }
 };
-// for decimal12_t
-template <>
-struct std::hash<doris::decimal12_t> {
-    int64_t operator()(const doris::decimal12_t& rhs) const {
-        return hash<int64_t>()(rhs.integer) ^ hash<int32_t>()(rhs.fraction);
-    }
-};
 
-template <>
-struct std::equal_to<doris::decimal12_t> {
-    bool operator()(const doris::decimal12_t& lhs, const doris::decimal12_t& rhs) const {
-        return lhs == rhs;
-    }
-};
 // for uint24_t
 template <>
 struct std::hash<doris::uint24_t> {
@@ -83,7 +70,7 @@ namespace doris {
 template <PrimitiveType Type, PredicateType PT, typename HybridSetType>
 class InListPredicateBase : public ColumnPredicate {
 public:
-    using T = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using T = typename PrimitiveTypeTraits<Type>::CppType;
     template <typename ConditionType, typename ConvertFunc>
     InListPredicateBase(uint32_t column_id, const ConditionType& conditions,
                         const ConvertFunc& convert, bool is_opposite = false,
@@ -134,9 +121,8 @@ class InListPredicateBase : public ColumnPredicate {
             } else if constexpr (Type == TYPE_DECIMALV2) {
                 HybridSetBase::IteratorBase* iter = hybrid_set->begin();
                 while (iter->has_next()) {
-                    const DecimalV2Value* value = (const DecimalV2Value*)(iter->get_value());
-                    decimal12_t decimal12 = {value->int_value(), value->frac_value()};
-                    _values->insert(&decimal12);
+                    const auto* value = (const DecimalV2Value*)(iter->get_value());
+                    _values->insert(value);
                     iter->next();
                 }
             } else if constexpr (Type == TYPE_DATE) {
@@ -329,18 +315,8 @@ class InListPredicateBase : public ColumnPredicate {
             return true;
         }
         if constexpr (PT == PredicateType::IN_LIST) {
-            if constexpr (Type == TYPE_DATE) {
-                T tmp_min_uint32_value = 0;
-                memcpy((char*)(&tmp_min_uint32_value), statistic.first->cell_ptr(),
-                       sizeof(uint24_t));
-                T tmp_max_uint32_value = 0;
-                memcpy((char*)(&tmp_max_uint32_value), statistic.second->cell_ptr(),
-                       sizeof(uint24_t));
-                return tmp_min_uint32_value <= _max_value && tmp_max_uint32_value >= _min_value;
-            } else {
-                return _get_zone_map_value<T>(statistic.first->cell_ptr()) <= _max_value &&
-                       _get_zone_map_value<T>(statistic.second->cell_ptr()) >= _min_value;
-            }
+            return get_zone_map_value<Type, T>(statistic.first->cell_ptr()) <= _max_value &&
+                   get_zone_map_value<Type, T>(statistic.second->cell_ptr()) >= _min_value;
         } else {
             return true;
         }
@@ -363,16 +339,16 @@ class InListPredicateBase : public ColumnPredicate {
         }
         if constexpr (PT == PredicateType::NOT_IN_LIST) {
             if constexpr (Type == TYPE_DATE) {
-                T tmp_min_uint32_value = 0;
+                T tmp_min_uint32_value {};
                 memcpy((char*)(&tmp_min_uint32_value), statistic.first->cell_ptr(),
                        sizeof(uint24_t));
-                T tmp_max_uint32_value = 0;
+                T tmp_max_uint32_value {};
                 memcpy((char*)(&tmp_max_uint32_value), statistic.second->cell_ptr(),
                        sizeof(uint24_t));
                 return tmp_min_uint32_value > _max_value || tmp_max_uint32_value < _min_value;
             } else {
-                return _get_zone_map_value<T>(statistic.first->cell_ptr()) > _max_value ||
-                       _get_zone_map_value<T>(statistic.second->cell_ptr()) < _min_value;
+                return get_zone_map_value<Type, T>(statistic.first->cell_ptr()) > _max_value ||
+                       get_zone_map_value<Type, T>(statistic.second->cell_ptr()) < _min_value;
             }
         } else {
             return false;
@@ -395,6 +371,16 @@ class InListPredicateBase : public ColumnPredicate {
                     if (bf->test_bytes(reinterpret_cast<const char*>(value), sizeof(uint24_t))) {
                         return true;
                     }
+                } else if constexpr (Type == PrimitiveType::TYPE_DECIMALV2) {
+                    // DecimalV2 using decimal12_t in bloom filter in storage layer,
+                    // should convert value to decimal12_t
+                    const T* value = (const T*)(iter->get_value());
+                    decimal12_t decimal12_t_val(value->int_value(), value->frac_value());
+                    if (bf->test_bytes(
+                                const_cast<char*>(reinterpret_cast<const char*>(&decimal12_t_val)),
+                                sizeof(decimal12_t))) {
+                        return true;
+                    }
                 } else {
                     const T* value = (const T*)(iter->get_value());
                     if (bf->test_bytes(reinterpret_cast<const char*>(value), sizeof(*value))) {
@@ -602,7 +588,7 @@ ColumnPredicate* _create_in_list_predicate(uint32_t column_id, const ConditionTy
                                            const ConvertFunc& convert, bool is_opposite = false,
                                            const TabletColumn* col = nullptr,
                                            vectorized::Arena* arena = nullptr) {
-    using T = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using T = typename PrimitiveTypeTraits<Type>::CppType;
     if constexpr (N >= 1 && N <= FIXED_CONTAINER_MAX_SIZE) {
         using Set = std::conditional_t<
                 std::is_same_v<T, StringRef>, StringSet<FixedContainer<std::string, N>>,
@@ -660,7 +646,7 @@ template <PrimitiveType Type, PredicateType PT, size_t N = 0>
 ColumnPredicate* _create_in_list_predicate(uint32_t column_id,
                                            const std::shared_ptr<HybridSetBase>& hybrid_set,
                                            size_t char_length = 0) {
-    using T = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using T = typename PrimitiveTypeTraits<Type>::CppType;
     if constexpr (N >= 1 && N <= FIXED_CONTAINER_MAX_SIZE) {
         using Set = std::conditional_t<
                 std::is_same_v<T, StringRef>, StringSet<FixedContainer<std::string, N>>,
diff --git a/be/src/olap/predicate_creator.h b/be/src/olap/predicate_creator.h
index 742336ec7774800..c47a1694f17a1d4 100644
--- a/be/src/olap/predicate_creator.h
+++ b/be/src/olap/predicate_creator.h
@@ -49,7 +49,7 @@ class PredicateCreator {
 template <PrimitiveType Type, PredicateType PT, typename ConditionType>
 class IntegerPredicateCreator : public PredicateCreator<ConditionType> {
 public:
-    using CppType = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using CppType = typename PrimitiveTypeTraits<Type>::CppType;
     ColumnPredicate* create(const TabletColumn& column, int index, const ConditionType& conditions,
                             bool opposite, vectorized::Arena* arena) override {
         if constexpr (PredicateTypeTraits::is_list(PT)) {
@@ -79,7 +79,7 @@ class IntegerPredicateCreator : public PredicateCreator<ConditionType> {
 template <PrimitiveType Type, PredicateType PT, typename ConditionType>
 class DecimalPredicateCreator : public PredicateCreator<ConditionType> {
 public:
-    using CppType = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using CppType = typename PrimitiveTypeTraits<Type>::CppType;
     ColumnPredicate* create(const TabletColumn& column, int index, const ConditionType& conditions,
                             bool opposite, vectorized::Arena* arena) override {
         if constexpr (PredicateTypeTraits::is_list(PT)) {
@@ -135,7 +135,7 @@ class StringPredicateCreator : public PredicateCreator<ConditionType> {
 template <PrimitiveType Type, PredicateType PT, typename ConditionType>
 struct CustomPredicateCreator : public PredicateCreator<ConditionType> {
 public:
-    using CppType = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using CppType = typename PrimitiveTypeTraits<Type>::CppType;
     CustomPredicateCreator(const std::function<CppType(const std::string& condition)>& convert)
             : _convert(convert) {}
 
@@ -183,7 +183,9 @@ std::unique_ptr<PredicateCreator<ConditionType>> get_creator(const FieldType& ty
                 [](const std::string& condition) {
                     decimal12_t value = {0, 0};
                     static_cast<void>(value.from_string(condition));
-                    return value;
+                    // Decimal12t is storage type, we need convert to compute type here to
+                    // do comparisons
+                    return DecimalV2Value(value.integer, value.fraction);
                 });
     }
     case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
diff --git a/be/src/olap/rowset/segment_v2/bitmap_index_reader.cpp b/be/src/olap/rowset/segment_v2/bitmap_index_reader.cpp
index c76de68b7ba33c6..3c6b1537169f1cf 100644
--- a/be/src/olap/rowset/segment_v2/bitmap_index_reader.cpp
+++ b/be/src/olap/rowset/segment_v2/bitmap_index_reader.cpp
@@ -58,7 +58,14 @@ Status BitmapIndexReader::new_iterator(BitmapIndexIterator** iterator) {
 }
 
 Status BitmapIndexIterator::seek_dictionary(const void* value, bool* exact_match) {
-    RETURN_IF_ERROR(_dict_column_iter.seek_at_or_after(value, exact_match));
+    auto st = _dict_column_iter.seek_at_or_after(value, exact_match);
+    if (!st.ok()) {
+        if (st.is<ErrorCode::ENTRY_NOT_FOUND>()) {
+            _current_rowid = 0;
+            return Status::OK();
+        }
+        return st;
+    }
     _current_rowid = _dict_column_iter.get_current_ordinal();
     return Status::OK();
 }
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 0140ba9cb57497f..1f0fabea7f5ea96 100644
--- a/be/src/olap/rowset/segment_v2/zone_map_index.cpp
+++ b/be/src/olap/rowset/segment_v2/zone_map_index.cpp
@@ -56,9 +56,7 @@ void TypedZoneMapIndexWriter<Type>::add_values(const void* values, size_t count)
     if (count > 0) {
         _page_zone_map.has_not_null = true;
     }
-    using ValType =
-            std::conditional_t<Type == TYPE_DATE, uint24_t,
-                               typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType>;
+    using ValType = PrimitiveTypeTraits<Type>::StorageFieldType;
     const ValType* vals = reinterpret_cast<const ValType*>(values);
     auto [min, max] = std::minmax_element(vals, vals + count);
     if (unaligned_load<ValType>(min) < unaligned_load<ValType>(_page_zone_map.min_value)) {
diff --git a/be/src/olap/types.h b/be/src/olap/types.h
index 40c68a3e369e3cf..5a263a8a39412a4 100644
--- a/be/src/olap/types.h
+++ b/be/src/olap/types.h
@@ -782,7 +782,7 @@ struct CppTypeTraits<FieldType::OLAP_FIELD_TYPE_MAP> {
     using CppType = MapValue;
 };
 template <FieldType field_type>
-struct BaseFieldtypeTraits : public CppTypeTraits<field_type> {
+struct BaseFieldTypeTraits : public CppTypeTraits<field_type> {
     using CppType = typename CppTypeTraits<field_type>::CppType;
 
     static inline CppType get_cpp_type_value(const void* address) {
@@ -846,7 +846,7 @@ struct BaseFieldtypeTraits : public CppTypeTraits<field_type> {
 // Using NumericFieldtypeTraits to Derived code for FieldType::OLAP_FIELD_TYPE_XXXINT, FieldType::OLAP_FIELD_TYPE_FLOAT,
 // FieldType::OLAP_FIELD_TYPE_DOUBLE, to reduce redundant code
 template <FieldType fieldType, bool isArithmetic>
-struct NumericFieldtypeTraits : public BaseFieldtypeTraits<fieldType> {
+struct NumericFieldTypeTraits : public BaseFieldTypeTraits<fieldType> {
     using CppType = typename CppTypeTraits<fieldType>::CppType;
 
     static std::string to_string(const void* src) {
@@ -855,19 +855,18 @@ struct NumericFieldtypeTraits : public BaseFieldtypeTraits<fieldType> {
 };
 
 template <FieldType fieldType>
-struct NumericFieldtypeTraits<fieldType, false> : public BaseFieldtypeTraits<fieldType> {};
+struct NumericFieldTypeTraits<fieldType, false> : public BaseFieldTypeTraits<fieldType> {};
 
 template <FieldType fieldType>
 struct FieldTypeTraits
-        : public NumericFieldtypeTraits<
+        : public NumericFieldTypeTraits<
                   fieldType,
-                  std::is_arithmetic<typename BaseFieldtypeTraits<fieldType>::CppType>::value &&
-                          std::is_signed<typename BaseFieldtypeTraits<fieldType>::CppType>::value> {
-};
+                  std::is_arithmetic_v<typename BaseFieldTypeTraits<fieldType>::CppType> &&
+                          std::is_signed_v<typename BaseFieldTypeTraits<fieldType>::CppType>> {};
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_BOOL>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_BOOL> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_BOOL> {
     static std::string to_string(const void* src) {
         char buf[1024] = {'\0'};
         snprintf(buf, sizeof(buf), "%d", *reinterpret_cast<const bool*>(src));
@@ -879,7 +878,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_BOOL>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_LARGEINT>
-        : public NumericFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_LARGEINT, true> {
+        : public NumericFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_LARGEINT, true> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         int128_t value = 0;
@@ -978,7 +977,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_LARGEINT>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_IPV4>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_IPV4> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_IPV4> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
@@ -1004,7 +1003,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_IPV4>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_IPV6>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_IPV6> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_IPV6> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         std::istringstream iss(scan_key);
@@ -1073,7 +1072,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_IPV6>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_FLOAT>
-        : public NumericFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_FLOAT, true> {
+        : public NumericFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_FLOAT, true> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         CppType value = 0.0f;
@@ -1095,7 +1094,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_FLOAT>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DOUBLE>
-        : public NumericFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DOUBLE, true> {
+        : public NumericFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DOUBLE, true> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         CppType value = 0.0;
@@ -1117,7 +1116,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DOUBLE>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         CppType* data_ptr = reinterpret_cast<CppType*>(buf);
@@ -1141,7 +1140,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL32>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL32> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL32> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
@@ -1159,7 +1158,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL32>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL64>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL64> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL64> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
@@ -1176,7 +1175,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL64>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL128I>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL128I> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL128I> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         StringParser::ParseResult result = StringParser::PARSE_SUCCESS;
@@ -1199,7 +1198,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL128I>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL256>
-        : public BaseFieldtypeTraits<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;
@@ -1226,7 +1225,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DECIMAL256>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATE>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DATE> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATE> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         tm time_tm;
@@ -1259,7 +1258,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATE>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATEV2>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DATEV2> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATEV2> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         tm time_tm;
@@ -1299,7 +1298,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATEV2>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIMEV2>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIMEV2> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIMEV2> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         DateV2Value<DateTimeV2ValueType> datetimev2_value;
@@ -1338,7 +1337,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIMEV2>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIME>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIME> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIME> {
     static Status from_string(void* buf, const std::string& scan_key, const int precision,
                               const int scale) {
         tm time_tm;
@@ -1385,7 +1384,7 @@ struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_DATETIME>
 
 template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_CHAR>
-        : public BaseFieldtypeTraits<FieldType::OLAP_FIELD_TYPE_CHAR> {
+        : public BaseFieldTypeTraits<FieldType::OLAP_FIELD_TYPE_CHAR> {
     static int cmp(const void* left, const void* right) {
         auto l_slice = reinterpret_cast<const Slice*>(left);
         auto r_slice = reinterpret_cast<const Slice*>(right);
diff --git a/be/src/runtime/decimalv2_value.h b/be/src/runtime/decimalv2_value.h
index 932ac01c0ba8083..aa9a2e2016001f8 100644
--- a/be/src/runtime/decimalv2_value.h
+++ b/be/src/runtime/decimalv2_value.h
@@ -327,3 +327,10 @@ template <>
 struct std::hash<doris::DecimalV2Value> {
     size_t operator()(const doris::DecimalV2Value& v) const { return doris::hash_value(v); }
 };
+
+template <>
+struct std::equal_to<doris::DecimalV2Value> {
+    bool operator()(const doris::DecimalV2Value& lhs, const doris::DecimalV2Value& rhs) const {
+        return lhs == rhs;
+    }
+};
\ No newline at end of file
diff --git a/be/src/runtime/primitive_type.h b/be/src/runtime/primitive_type.h
index 06e6842363b8894..8d97dd804253932 100644
--- a/be/src/runtime/primitive_type.h
+++ b/be/src/runtime/primitive_type.h
@@ -122,168 +122,167 @@ constexpr PrimitiveType PredicateEvaluateType = is_variant_string_type(type) ? T
 template <PrimitiveType type>
 struct PrimitiveTypeTraits;
 
+/// CppType as type on compute layer
+/// StorageFieldType as type on storage layer
 template <>
 struct PrimitiveTypeTraits<TYPE_BOOLEAN> {
     using CppType = bool;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnUInt8;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_TINYINT> {
     using CppType = int8_t;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnInt8;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_SMALLINT> {
     using CppType = int16_t;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnInt16;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_INT> {
     using CppType = int32_t;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnInt32;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_BIGINT> {
     using CppType = int64_t;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnInt64;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_FLOAT> {
     using CppType = float;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnFloat32;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_TIME> {
     using CppType = double;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnFloat64;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_TIMEV2> {
     using CppType = double;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnFloat64;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DOUBLE> {
     using CppType = double;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnFloat64;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DATE> {
     using CppType = doris::VecDateTimeValue;
+    /// Different with compute layer, the DateV1 was stored as uint24_t(3 bytes).
+    using StorageFieldType = uint24_t;
     using ColumnType = vectorized::ColumnVector<vectorized::Int64>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DATETIME> {
     using CppType = doris::VecDateTimeValue;
+    using StorageFieldType = uint64_t;
     using ColumnType = vectorized::ColumnVector<vectorized::Int64>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DATETIMEV2> {
     using CppType = DateV2Value<DateTimeV2ValueType>;
+    using StorageFieldType = uint64_t;
     using ColumnType = vectorized::ColumnVector<vectorized::UInt64>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DATEV2> {
     using CppType = DateV2Value<DateV2ValueType>;
+    using StorageFieldType = uint32_t;
     using ColumnType = vectorized::ColumnVector<vectorized::UInt32>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DECIMALV2> {
     using CppType = DecimalV2Value;
+    /// Different with compute layer, the DecimalV1 was stored as decimal12_t(12 bytes).
+    using StorageFieldType = decimal12_t;
     using ColumnType = vectorized::ColumnDecimal<vectorized::Decimal128>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DECIMAL32> {
     using CppType = vectorized::Decimal32;
+    using StorageFieldType = vectorized::Int32;
     using ColumnType = vectorized::ColumnDecimal<vectorized::Decimal32>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DECIMAL64> {
     using CppType = vectorized::Decimal64;
+    using StorageFieldType = vectorized::Int64;
     using ColumnType = vectorized::ColumnDecimal<vectorized::Decimal64>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DECIMAL128I> {
     using CppType = vectorized::Decimal128I;
+    using StorageFieldType = vectorized::Int128;
     using ColumnType = vectorized::ColumnDecimal<vectorized::Decimal128I>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_DECIMAL256> {
     using CppType = vectorized::Decimal256;
+    using StorageFieldType = Int256;
     using ColumnType = vectorized::ColumnDecimal<vectorized::Decimal256>;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_LARGEINT> {
     using CppType = __int128_t;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnInt128;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_IPV4> {
     using CppType = vectorized::IPv4;
+    using StorageFieldType = uint32_t;
     using ColumnType = vectorized::ColumnIPv4;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_IPV6> {
     using CppType = vectorized::IPv6;
+    using StorageFieldType = uint64_t;
     using ColumnType = vectorized::ColumnIPv6;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_CHAR> {
     using CppType = StringRef;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnString;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_VARCHAR> {
     using CppType = StringRef;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnString;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_STRING> {
     using CppType = StringRef;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnString;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_HLL> {
     using CppType = StringRef;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnString;
 };
 template <>
 struct PrimitiveTypeTraits<TYPE_JSONB> {
     using CppType = JsonBinaryValue;
+    using StorageFieldType = CppType;
     using ColumnType = vectorized::ColumnString;
 };
 
-// only for adapt get_predicate_column_ptr
-template <PrimitiveType type>
-struct PredicatePrimitiveTypeTraits {
-    using PredicateFieldType = typename PrimitiveTypeTraits<type>::CppType;
-};
-
-template <>
-struct PredicatePrimitiveTypeTraits<TYPE_DECIMALV2> {
-    using PredicateFieldType = decimal12_t;
-};
-
-template <>
-struct PredicatePrimitiveTypeTraits<TYPE_DATE> {
-    using PredicateFieldType = uint32_t;
-};
-
-template <>
-struct PredicatePrimitiveTypeTraits<TYPE_DATETIME> {
-    using PredicateFieldType = uint64_t;
-};
-
-template <>
-struct PredicatePrimitiveTypeTraits<TYPE_DATEV2> {
-    using PredicateFieldType = uint32_t;
-};
-
-template <>
-struct PredicatePrimitiveTypeTraits<TYPE_DATETIMEV2> {
-    using PredicateFieldType = uint64_t;
-};
-
 template <typename Traits>
 concept HaveCppType = requires() { sizeof(typename Traits::CppType); };
 
diff --git a/be/src/util/date_func.cpp b/be/src/util/date_func.cpp
index a9198a249ab65ce..d3ad3e0a1be1112 100644
--- a/be/src/util/date_func.cpp
+++ b/be/src/util/date_func.cpp
@@ -28,7 +28,7 @@
 
 namespace doris {
 
-uint64_t timestamp_from_datetime(const std::string& datetime_str) {
+VecDateTimeValue timestamp_from_datetime(const std::string& datetime_str) {
     tm time_tm;
     char* res = strptime(datetime_str.c_str(), "%Y-%m-%d %H:%M:%S", &time_tm);
 
@@ -43,10 +43,10 @@ uint64_t timestamp_from_datetime(const std::string& datetime_str) {
         value = 14000101000000;
     }
 
-    return value;
+    return VecDateTimeValue::create_from_olap_datetime(value);
 }
 
-uint32_t timestamp_from_date(const std::string& date_str) {
+VecDateTimeValue timestamp_from_date(const std::string& date_str) {
     tm time_tm;
     char* res = strptime(date_str.c_str(), "%Y-%m-%d", &time_tm);
 
@@ -60,10 +60,10 @@ uint32_t timestamp_from_date(const std::string& date_str) {
         value = 716833;
     }
 
-    return value;
+    return VecDateTimeValue::create_from_olap_date(value);
 }
 
-uint32_t timestamp_from_date_v2(const std::string& date_str) {
+DateV2Value<DateV2ValueType> timestamp_from_date_v2(const std::string& date_str) {
     tm time_tm;
     char* res = strptime(date_str.c_str(), "%Y-%m-%d", &time_tm);
 
@@ -74,15 +74,15 @@ uint32_t timestamp_from_date_v2(const std::string& date_str) {
         value = MIN_DATE_V2;
     }
 
-    return value;
+    return DateV2Value<DateV2ValueType>::create_from_olap_date(value);
 }
 
-uint64_t timestamp_from_datetime_v2(const std::string& date_str) {
+DateV2Value<DateTimeV2ValueType> timestamp_from_datetime_v2(const std::string& date_str) {
     DateV2Value<DateTimeV2ValueType> val;
     std::string date_format = "%Y-%m-%d %H:%i:%s.%f";
     val.from_date_format_str(date_format.data(), date_format.size(), date_str.data(),
                              date_str.size());
-    return val.to_date_int_val();
+    return val;
 }
 // refer to https://dev.mysql.com/doc/refman/5.7/en/time.html
 // the time value between '-838:59:59' and '838:59:59'
diff --git a/be/src/util/date_func.h b/be/src/util/date_func.h
index 405132c3f01282d..dbf8a738e7819ab 100644
--- a/be/src/util/date_func.h
+++ b/be/src/util/date_func.h
@@ -23,12 +23,20 @@
 
 namespace doris {
 
-uint64_t timestamp_from_datetime(const std::string& datetime_str);
-uint32_t timestamp_from_date(const std::string& date_str);
+struct DateV2ValueType;
+struct DateTimeV2ValueType;
+
+template <typename T>
+class DateV2Value;
+
+class VecDateTimeValue;
+
+VecDateTimeValue timestamp_from_datetime(const std::string& datetime_str);
+VecDateTimeValue timestamp_from_date(const std::string& date_str);
 int32_t time_to_buffer_from_double(double time, char* buffer);
 int32_t timev2_to_buffer_from_double(double time, char* buffer, int scale);
-uint32_t timestamp_from_date_v2(const std::string& date_str);
-uint64_t timestamp_from_datetime_v2(const std::string& date_str);
+DateV2Value<DateV2ValueType> timestamp_from_date_v2(const std::string& date_str);
+DateV2Value<DateTimeV2ValueType> timestamp_from_datetime_v2(const std::string& date_str);
 
 std::string time_to_buffer_from_double(double time);
 std::string timev2_to_buffer_from_double(double time, int scale);
diff --git a/be/src/vec/columns/predicate_column.h b/be/src/vec/columns/predicate_column.h
index 3aec3b1540f2294..c4cd53f0737779d 100644
--- a/be/src/vec/columns/predicate_column.h
+++ b/be/src/vec/columns/predicate_column.h
@@ -45,29 +45,26 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
     PredicateColumnType(const size_t n) : data(n) {}
     PredicateColumnType(const PredicateColumnType& src) : data(src.data.begin(), src.data.end()) {}
     friend class COWHelper<IColumn, PredicateColumnType<Type>>;
-    using T = typename PredicatePrimitiveTypeTraits<Type>::PredicateFieldType;
+    using T = typename PrimitiveTypeTraits<Type>::CppType;
     using ColumnType = typename PrimitiveTypeTraits<Type>::ColumnType;
 
-    void insert_date_to_res_column(const uint16_t* sel, size_t sel_size,
-                                   ColumnVector<Int64>* res_ptr) {
-        res_ptr->reserve(sel_size);
-        auto& res_data = res_ptr->get_data();
-
+    void insert_datetime_to_res_column(const uint16_t* sel, size_t sel_size,
+                                       ColumnVector<Int64>* res) {
+        auto& res_data = res->get_data();
         for (size_t i = 0; i < sel_size; i++) {
-            uint64_t val = data[sel[i]];
-            VecDateTimeValue date;
-            date.set_olap_date(val);
-            res_data.push_back_without_reserve(
-                    unaligned_load<Int64>(reinterpret_cast<char*>(&date)));
+            res_data.push_back(*reinterpret_cast<int64_t*>(&data[sel[i]]));
         }
     }
 
-    void insert_datetime_to_res_column(const uint16_t* sel, size_t sel_size,
-                                       ColumnVector<Int64>* res_ptr) {
+    void insert_datev2_to_res_column(const uint16_t* sel, size_t sel_size, ColumnType* res_ptr) {
+        res_ptr->reserve(sel_size);
+        auto& res_data = res_ptr->get_data();
+        DCHECK_EQ(sizeof(T), sizeof(res_data[0]));
+
         for (size_t i = 0; i < sel_size; i++) {
-            uint64_t value = data[sel[i]];
-            VecDateTimeValue datetime = VecDateTimeValue::create_from_olap_datetime(value);
-            res_ptr->insert_data(reinterpret_cast<char*>(&datetime), 0);
+            const auto value = data[sel[i]];
+            res_data.push_back_without_reserve(
+                    unaligned_load<Int64>(reinterpret_cast<const char*>(&value)));
         }
     }
 
@@ -86,16 +83,6 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
         res_ptr->insert_many_strings_without_reserve(refs, sel_size);
     }
 
-    void insert_decimal_to_res_column(const uint16_t* sel, size_t sel_size,
-                                      ColumnDecimal<Decimal128>* res_ptr) {
-        for (size_t i = 0; i < sel_size; i++) {
-            uint16_t n = sel[i];
-            auto& dv = reinterpret_cast<const decimal12_t&>(data[n]);
-            DecimalV2Value dv_data(dv.integer, dv.fraction);
-            res_ptr->insert_data(reinterpret_cast<char*>(&dv_data), 0);
-        }
-    }
-
     template <typename Y, template <typename> typename ColumnContainer>
     void insert_default_value_res_column(const uint16_t* sel, size_t sel_size,
                                          ColumnContainer<Y>* res_ptr) {
@@ -192,7 +179,7 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
         decimal12_t dc12_value;
         dc12_value.integer = *(int64_t*)(data_ptr);
         dc12_value.fraction = *(int32_t*)(data_ptr + sizeof(int64_t));
-        data.push_back_without_reserve(dc12_value);
+        data.push_back_without_reserve(DecimalV2Value(dc12_value.integer, dc12_value.fraction));
     }
 
     // used for int128
@@ -220,29 +207,63 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
     }
 
     void insert_many_date(const char* data_ptr, size_t num) {
-        size_t intput_type_size = sizeof(uint24_t);
-        size_t res_type_size = sizeof(uint32_t);
+        constexpr size_t input_type_size = sizeof(PrimitiveTypeTraits<TYPE_DATE>::StorageFieldType);
+        static_assert(input_type_size == sizeof(uint24_t));
         char* input_data_ptr = const_cast<char*>(data_ptr);
 
-        char* res_ptr = (char*)data.get_end_ptr();
-        memset(res_ptr, 0, res_type_size * num);
+        auto* res_ptr = reinterpret_cast<VecDateTimeValue*>(data.get_end_ptr());
         for (int i = 0; i < num; i++) {
-            memcpy(res_ptr, input_data_ptr, intput_type_size);
-            res_ptr += res_type_size;
-            input_data_ptr += intput_type_size;
+            uint64_t value = *reinterpret_cast<uint24_t*>(input_data_ptr);
+            res_ptr[i].set_olap_date(value);
+            input_data_ptr += input_type_size;
+        }
+        data.set_end_ptr(res_ptr + num);
+    }
+
+    void insert_many_datetime(const char* data_ptr, size_t num) {
+        constexpr size_t input_type_size =
+                sizeof(PrimitiveTypeTraits<TYPE_DATETIME>::StorageFieldType);
+        static_assert(input_type_size == sizeof(uint64_t));
+        char* input_data_ptr = const_cast<char*>(data_ptr);
+
+        auto* res_ptr = reinterpret_cast<VecDateTimeValue*>(data.get_end_ptr());
+        for (int i = 0; i < num; i++) {
+            res_ptr[i].from_olap_datetime(*reinterpret_cast<uint64_t*>(input_data_ptr));
+            input_data_ptr += input_type_size;
+        }
+        data.set_end_ptr(res_ptr + num);
+    }
+
+    // The logic is same to ColumnDecimal::insert_many_fix_len_data
+    void insert_many_decimalv2(const char* data_ptr, size_t num) {
+        size_t old_size = data.size();
+        data.resize(old_size + num);
+
+        DecimalV2Value* target = (DecimalV2Value*)(data.data() + old_size);
+        for (int i = 0; i < num; i++) {
+            const char* cur_ptr = data_ptr + sizeof(decimal12_t) * i;
+            int64_t int_value = unaligned_load<int64_t>(cur_ptr);
+            int32_t frac_value = *(int32_t*)(cur_ptr + sizeof(int64_t));
+            target[i].from_olap_decimal(int_value, frac_value);
         }
-        data.set_end_ptr(res_ptr);
     }
 
     void insert_many_fix_len_data(const char* data_ptr, size_t num) override {
-        if constexpr (std::is_same_v<T, decimal12_t>) {
-            insert_many_in_copy_way(data_ptr, num);
+        if constexpr (Type == TYPE_DECIMALV2) {
+            // DecimalV2 is special, its storage is <int64, int32>, but its compute type is <int64,int64>
+            // should convert here, but it may have some performance lost
+            insert_many_decimalv2(data_ptr, num);
         } else if constexpr (std::is_same_v<T, Int128>) {
+            // Int128 need copy to ensure alignment, or the code may core during SIMD
             insert_many_in_copy_way(data_ptr, num);
         } else if constexpr (std::is_same_v<T, StringRef>) {
             // here is unreachable, just for compilation to be able to pass
         } else if constexpr (Type == TYPE_DATE) {
+            // Datev1 is special, its storage is uint24, but its compute type is uint32, we will change
+            // to its actual type int64, in the future.
             insert_many_date(data_ptr, num);
+        } else if constexpr (Type == TYPE_DATETIME) {
+            insert_many_datetime(data_ptr, num);
         } else {
             insert_many_default_type(data_ptr, num);
         }
@@ -450,13 +471,9 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
         // DateV1 and DateTimeV1 is special, its storage format is different from compute format
         // should convert here.
         if constexpr (Type == TYPE_DATE || Type == TYPE_DATETIME) {
-            if constexpr (std::is_same_v<T, uint32_t>) {
-                insert_date_to_res_column(sel, sel_size, column);
-            } else if constexpr (std::is_same_v<T, uint64_t>) {
-                insert_datetime_to_res_column(sel, sel_size, column);
-            } else {
-                LOG(FATAL) << "not reachable";
-            }
+            insert_datetime_to_res_column(sel, sel_size, column);
+        } else if constexpr (Type == TYPE_DATEV2 || Type == TYPE_DATETIMEV2) {
+            insert_datev2_to_res_column(sel, sel_size, column);
         } else if constexpr (std::is_same_v<ColumnVector<T>, ColumnType>) {
             // DateV2 and DateTimeV2, its storage format is equal to compute format
             // not need convert
@@ -465,8 +482,6 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
             insert_default_value_res_column(sel, sel_size, column);
         } else if constexpr (std::is_same_v<T, StringRef>) {
             insert_string_to_res_column(sel, sel_size, column);
-        } else if constexpr (std::is_same_v<T, decimal12_t>) {
-            insert_decimal_to_res_column(sel, sel_size, column);
         } else if (std::is_same_v<T, bool>) {
             insert_byte_to_res_column(sel, sel_size, col_ptr);
         } else if constexpr (std::is_same_v<T, doris::vectorized::IPv4>) {
diff --git a/be/src/vec/core/types.h b/be/src/vec/core/types.h
index 861126a030ed42e..9ddaa598580620d 100644
--- a/be/src/vec/core/types.h
+++ b/be/src/vec/core/types.h
@@ -228,6 +228,26 @@ struct TypeName<QuantileState> {
     static const char* get() { return "QuantileState"; }
 };
 
+template <>
+struct TypeName<DecimalV2Value> {
+    static const char* get() { return "decimalv2"; }
+};
+
+template <>
+struct TypeName<VecDateTimeValue> {
+    static const char* get() { return "Datetime"; }
+};
+
+template <>
+struct TypeName<DateV2Value<DateV2ValueType>> {
+    static const char* get() { return "DateV2"; }
+};
+
+template <>
+struct TypeName<DateV2Value<DateTimeV2ValueType>> {
+    static const char* get() { return "DatetimeV2"; }
+};
+
 template <typename T>
 struct TypeId;
 template <>
diff --git a/be/test/util/date_func_test.cpp b/be/test/util/date_func_test.cpp
index c29dabfe62ec843..4e46accb5e07db0 100644
--- a/be/test/util/date_func_test.cpp
+++ b/be/test/util/date_func_test.cpp
@@ -22,6 +22,7 @@
 
 #include "gtest/gtest_pred_impl.h"
 #include "olap/uint24.h"
+#include "vec/runtime/vdatetime_value.h"
 
 namespace doris {
 
@@ -32,16 +33,18 @@ class DateFuncTest : public testing::Test {
 };
 
 TEST_F(DateFuncTest, convert_string_to_int) {
-    uint64_t result1 = timestamp_from_datetime(std::string("2021-06-08 15:21:18"));
+    uint64_t result1 =
+            timestamp_from_datetime(std::string("2021-06-08 15:21:18")).to_olap_datetime();
     EXPECT_EQ(20210608152118, result1);
 
-    uint64_t abnormal_result1 = timestamp_from_datetime(std::string("2021-22-08 15:21:18"));
+    uint64_t abnormal_result1 =
+            timestamp_from_datetime(std::string("2021-22-08 15:21:18")).to_olap_datetime();
     EXPECT_EQ(14000101000000, abnormal_result1);
 
-    uint24_t result2 = timestamp_from_date(std::string("2021-09-08"));
+    uint24_t result2 = timestamp_from_date(std::string("2021-09-08")).to_olap_date();
     EXPECT_EQ(std::string("2021-09-08"), result2.to_string());
 
-    uint24_t abnormal_result2 = timestamp_from_date(std::string("2021-25-08"));
+    uint24_t abnormal_result2 = timestamp_from_date(std::string("2021-25-08")).to_olap_date();
     EXPECT_EQ(std::string("1400-01-01"), abnormal_result2.to_string());
 }
 
diff --git a/regression-test/data/datatype_p0/scalar_types/sql/infomation_schema.out b/regression-test/data/datatype_p0/scalar_types/sql/infomation_schema.out
index cceb1c0f8587d29..a5c5eaad199399f 100644
--- a/regression-test/data/datatype_p0/scalar_types/sql/infomation_schema.out
+++ b/regression-test/data/datatype_p0/scalar_types/sql/infomation_schema.out
@@ -12,7 +12,7 @@ internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_double
 internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_decimal	10	\N	YES	decimal	\N	\N	20	3	\N	\N	\N	decimalv3(20, 3)					20	3	\N	\N
 internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_decimalv3	11	\N	YES	decimal	\N	\N	20	3	\N	\N	\N	decimalv3(20, 3)					20	3	\N	\N
 internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_date	12	\N	YES	date	\N	\N	\N	\N	\N	\N	\N	date					\N	\N	\N	\N
-internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_datetime	13	\N	YES	datetime	\N	\N	18	0	\N	\N	\N	datetime					\N	0	\N	\N
+internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_datetime	13	\N	YES	datetime	\N	\N	\N	\N	\N	\N	\N	datetime					\N	\N	\N	\N
 internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_datev2	14	\N	YES	date	\N	\N	\N	\N	\N	\N	\N	date					\N	\N	\N	\N
 internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_datetimev2	15	\N	YES	datetime	\N	\N	18	0	\N	\N	\N	datetime					\N	0	\N	\N
 internal	regression_test_datatype_p0_scalar_types	tbl_scalar_types_dup	c_char	16	\N	YES	char	15	60	\N	\N	\N	\N	\N	char(15)					15	\N	\N	\N
diff --git a/regression-test/suites/datatype_p0/scalar_types/load.groovy b/regression-test/suites/datatype_p0/scalar_types/load.groovy
index 7577bbb819ce8ea..f360097b98bfa9a 100644
--- a/regression-test/suites/datatype_p0/scalar_types/load.groovy
+++ b/regression-test/suites/datatype_p0/scalar_types/load.groovy
@@ -37,8 +37,8 @@ suite("test_scalar_types_load", "p0") {
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
             `c_decimalv3` decimalv3(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_datetimev2` datetimev2(0) NULL,
             `c_char` char(15) NULL,
@@ -90,8 +90,8 @@ suite("test_scalar_types_load", "p0") {
             `c_float` float NULL,
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_char` char(15) NULL,
             `c_varchar` varchar(100) NULL,
@@ -134,8 +134,8 @@ suite("test_scalar_types_load", "p0") {
             `c_float` float NULL,
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_char` char(15) NULL,
             `c_varchar` varchar(100) NULL,
@@ -174,8 +174,8 @@ suite("test_scalar_types_load", "p0") {
             `c_float` float NULL,
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_char` char(15) NULL,
             `c_varchar` varchar(100) NULL,
@@ -214,8 +214,8 @@ suite("test_scalar_types_load", "p0") {
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
             `c_decimalv3` decimalv3(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_datetimev2` datetimev2(0) NULL,
             `c_char` char(15) NULL,
@@ -264,8 +264,8 @@ suite("test_scalar_types_load", "p0") {
             `c_float` float NULL,
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_char` char(15) NULL,
             `c_varchar` varchar(100) NULL,
@@ -319,8 +319,8 @@ suite("test_scalar_types_load", "p0") {
             `c_float` float NULL,
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_char` char(15) NULL,
             `c_varchar` varchar(100) NULL,
@@ -374,8 +374,8 @@ suite("test_scalar_types_load", "p0") {
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
             `c_decimalv3` decimalv3(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_datetimev2` datetimev2(0) NULL,
             `c_char` char(15) NULL,
@@ -423,8 +423,8 @@ suite("test_scalar_types_load", "p0") {
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
             `c_decimalv3` decimalv3(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_datetimev2` datetimev2(0) NULL,
             `c_char` char(15) NULL,
@@ -472,8 +472,8 @@ suite("test_scalar_types_load", "p0") {
             `c_float` float NULL,
             `c_double` double NULL,
             `c_decimal` decimal(20, 3) NULL,
-            `c_date` date NULL,
-            `c_datetime` datetime NULL,
+            `c_date` datev1 NULL,
+            `c_datetime` datetimev1 NULL,
             `c_datev2` datev2 NULL,
             `c_char` char(15) NULL,
             `c_varchar` varchar(100) NULL,