diff --git a/be/src/exprs/bloom_filter_func.h b/be/src/exprs/bloom_filter_func.h
index a8330250ec084d..dfb775cc0a8e8a 100644
--- a/be/src/exprs/bloom_filter_func.h
+++ b/be/src/exprs/bloom_filter_func.h
@@ -416,65 +416,12 @@ 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);
-
-        constexpr int decimal_value_sz = sizeof(DecimalV2Value);
-        char data_bytes[decimal_value_sz];
-        memcpy(&data_bytes, &value, decimal_value_sz);
-        return bloom_filter.test(Slice(data_bytes, decimal_value_sz));
-    }
-};
-
 template <PrimitiveType type>
 struct BloomFilterTypeTraits {
     using T = typename PrimitiveTypeTraits<type>::CppType;
     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;
-};
-
 template <>
 struct BloomFilterTypeTraits<TYPE_CHAR> {
     using FindOp = FixedStringFindOp;
diff --git a/be/src/olap/bitmap_filter_predicate.h b/be/src/olap/bitmap_filter_predicate.h
index f90e0e625b8611..de03a1bc61a45e 100644
--- a/be/src/olap/bitmap_filter_predicate.h
+++ b/be/src/olap/bitmap_filter_predicate.h
@@ -58,13 +58,13 @@ class BitmapFilterColumnPredicate : public ColumnPredicate {
             // no non-null values
             return false;
         } else {
-            max_value = *reinterpret_cast<const CppType*>(statistic.second->cell_ptr());
+            max_value = get_zone_map_value<T, CppType>(statistic.second->cell_ptr());
         }
 
-        CppType min_value =
-                statistic.first->is_null() /* contains null values */
-                        ? 0
-                        : *reinterpret_cast<const CppType*>(statistic.first->cell_ptr());
+        CppType min_value = statistic.first->is_null() /* contains null values */
+                                    ? 0
+                                    : get_zone_map_value<T, CppType>(statistic.first->cell_ptr());
+        ;
         return _specific_filter->contains_any(min_value, max_value);
     }
 
diff --git a/be/src/olap/bloom_filter_predicate.h b/be/src/olap/bloom_filter_predicate.h
index d2816be9966dc3..7280a1e836254c 100644
--- a/be/src/olap/bloom_filter_predicate.h
+++ b/be/src/olap/bloom_filter_predicate.h
@@ -63,20 +63,9 @@ class BloomFilterColumnPredicate : public ColumnPredicate {
             DCHECK(null_map);
         }
 
-        uint24_t tmp_uint24_value;
-        auto get_cell_value = [&tmp_uint24_value](auto& data) {
-            if constexpr (std::is_same_v<std::decay_t<decltype(data)>, uint32_t> &&
-                          T == PrimitiveType::TYPE_DATE) {
-                memcpy((char*)(&tmp_uint24_value), (char*)(&data), sizeof(uint24_t));
-                return (const char*)&tmp_uint24_value;
-            } else {
-                return (const char*)&data;
-            }
-        };
-
         uint16_t new_size = 0;
         if (column.is_column_dictionary()) {
-            auto* dict_col = reinterpret_cast<const vectorized::ColumnDictI32*>(&column);
+            const auto* dict_col = reinterpret_cast<const vectorized::ColumnDictI32*>(&column);
             if (_be_exec_version >= 2) {
                 for (uint16_t i = 0; i < size; i++) {
                     uint16_t idx = sel[i];
@@ -113,12 +102,11 @@ class BloomFilterColumnPredicate : public ColumnPredicate {
             for (uint16_t i = 0; i < size; i++) {
                 uint16_t idx = is_dense_column ? i : sel[i];
                 if constexpr (is_nullable) {
-                    if (!null_map[idx] &&
-                        _specific_filter->find_crc32_hash(get_cell_value(pred_col_data[idx]))) {
+                    if (!null_map[idx] && _specific_filter->find_crc32_hash(&pred_col_data[idx])) {
                         sel[new_size++] = idx;
                     }
                 } else {
-                    if (_specific_filter->find_crc32_hash(get_cell_value(pred_col_data[idx]))) {
+                    if (_specific_filter->find_crc32_hash(&pred_col_data[idx])) {
                         sel[new_size++] = idx;
                     }
                 }
@@ -140,9 +128,8 @@ class BloomFilterColumnPredicate : public ColumnPredicate {
 
             auto pred_col_data = pred_col.data();
 #define EVALUATE_WITH_NULL_IMPL(IDX) \
-    !null_map[IDX] && _specific_filter->find_olap_engine(get_cell_value(pred_col_data[IDX]))
-#define EVALUATE_WITHOUT_NULL_IMPL(IDX) \
-    _specific_filter->find_olap_engine(get_cell_value(pred_col_data[IDX]))
+    !null_map[IDX] && _specific_filter->find_olap_engine(&pred_col_data[IDX])
+#define EVALUATE_WITHOUT_NULL_IMPL(IDX) _specific_filter->find_olap_engine(&pred_col_data[IDX])
             EVALUATE_BY_SELECTOR(EVALUATE_WITH_NULL_IMPL, EVALUATE_WITHOUT_NULL_IMPL)
 #undef EVALUATE_WITH_NULL_IMPL
 #undef EVALUATE_WITHOUT_NULL_IMPL
diff --git a/be/src/olap/column_predicate.h b/be/src/olap/column_predicate.h
index 05e84999a83102..15b36f672eaf6f 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(reinterpret_cast<void*>(&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 522939370cf444..1c2377477c7126 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,9 @@ class ComparisonPredicateBase : public ColumnPredicate {
 
         roaring::Roaring roaring;
         bool exact_match = false;
-        Status status = iterator->seek_dictionary(&_value, &exact_match);
+
+        auto&& value = PrimitiveTypeConvertor<Type>::to_storage_field_type(_value);
+        Status 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,7 +109,9 @@ class ComparisonPredicateBase : public ColumnPredicate {
         }
 
         roaring::Roaring roaring;
-        RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, &_value, query_type,
+
+        auto&& value = PrimitiveTypeConvertor<Type>::to_storage_field_type(_value);
+        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
@@ -150,17 +154,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 +183,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 +204,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 +228,17 @@ 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
+                // Datev1/DatetimeV1 using VecDatetimeValue in bloom filter, NO need to convert.
+                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 ad4a4e8c8760a3..35d4a559e2b9c6 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 cb733ab9b650bf..734c7d75bd4949 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,
@@ -131,30 +118,6 @@ class InListPredicateBase : public ColumnPredicate {
                     }
                     iter->next();
                 }
-            } 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);
-                    iter->next();
-                }
-            } else if constexpr (Type == TYPE_DATE) {
-                HybridSetBase::IteratorBase* iter = hybrid_set->begin();
-                while (iter->has_next()) {
-                    const VecDateTimeValue* value = (const VecDateTimeValue*)(iter->get_value());
-                    uint64_t date = value->to_olap_date();
-                    _values->insert(&date);
-                    iter->next();
-                }
-            } else if constexpr (Type == TYPE_DATETIME) {
-                HybridSetBase::IteratorBase* iter = hybrid_set->begin();
-                while (iter->has_next()) {
-                    const VecDateTimeValue* value = (const VecDateTimeValue*)(iter->get_value());
-                    uint64_t date_time = value->to_olap_datetime();
-                    _values->insert(&date_time);
-                    iter->next();
-                }
             } else {
                 HybridSetBase::IteratorBase* iter = hybrid_set->begin();
                 while (iter->has_next()) {
@@ -162,7 +125,6 @@ class InListPredicateBase : public ColumnPredicate {
                     _values->insert(value);
                     iter->next();
                 }
-                CHECK(Type == TYPE_DATETIMEV2 || Type == TYPE_DATEV2);
             }
         } else {
             // shared from the caller, so it needs to be shared ptr
@@ -195,11 +157,13 @@ class InListPredicateBase : public ColumnPredicate {
         while (iter->has_next()) {
             const void* value = iter->get_value();
             bool exact_match;
-            Status s = iterator->seek_dictionary(value, &exact_match);
+            auto&& value_ = PrimitiveTypeConvertor<Type>::to_storage_field_type(
+                    *reinterpret_cast<const T*>(value));
+            Status status = iterator->seek_dictionary(&value_, &exact_match);
             rowid_t seeked_ordinal = iterator->current_ordinal();
-            if (!s.is<ErrorCode::ENTRY_NOT_FOUND>()) {
-                if (!s.ok()) {
-                    return s;
+            if (!status.is<ErrorCode::ENTRY_NOT_FOUND>()) {
+                if (!status.ok()) {
+                    return status;
                 }
                 if (exact_match) {
                     roaring::Roaring index;
@@ -229,10 +193,12 @@ class InListPredicateBase : public ColumnPredicate {
         roaring::Roaring indices;
         HybridSetBase::IteratorBase* iter = _values->begin();
         while (iter->has_next()) {
-            const void* value = iter->get_value();
+            const void* ptr = iter->get_value();
+            auto&& value = PrimitiveTypeConvertor<Type>::to_storage_field_type(
+                    *reinterpret_cast<const T*>(ptr));
             InvertedIndexQueryType query_type = InvertedIndexQueryType::EQUAL_QUERY;
             roaring::Roaring index;
-            RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, value, query_type,
+            RETURN_IF_ERROR(iterator->read_from_inverted_index(column_name, &value, query_type,
                                                                num_rows, &index));
             indices |= index;
             iter->next();
@@ -329,18 +295,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;
         }
@@ -362,18 +318,8 @@ class InListPredicateBase : public ColumnPredicate {
             return false;
         }
         if constexpr (PT == PredicateType::NOT_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 false;
         }
@@ -390,9 +336,15 @@ class InListPredicateBase : public ColumnPredicate {
                     if (bf->test_bytes(value->data, value->size)) {
                         return true;
                     }
-                } else if constexpr (Type == TYPE_DATE) {
-                    const void* value = iter->get_value();
-                    if (bf->test_bytes(reinterpret_cast<const char*>(value), sizeof(uint24_t))) {
+                } else if constexpr (Type == PrimitiveType::TYPE_DECIMALV2) {
+                    // DecimalV2 using decimal12_t in bloom filter in storage layer,
+                    // should convert value to decimal12_t
+                    // Datev1/DatetimeV1 using VecDatetimeValue in bloom filter, NO need to convert.
+                    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 {
@@ -602,7 +554,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 +612,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 742336ec777480..c47a1694f17a1d 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/zone_map_index.cpp b/be/src/olap/rowset/segment_v2/zone_map_index.cpp
index 0140ba9cb57497..1f0fabea7f5ea9 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 a0c38975f72978..b01cee2484353e 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;
@@ -1221,7 +1220,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;
@@ -1254,7 +1253,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;
@@ -1294,7 +1293,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;
@@ -1333,7 +1332,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;
@@ -1380,7 +1379,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 932ac01c0ba808..aa9a2e2016001f 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 06e6842363b889..4c36ac21bff296 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 = wide::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); };
 
@@ -308,4 +307,48 @@ inline size_t get_primitive_type_size(PrimitiveType t) {
     return size;
 }
 
+template <PrimitiveType PT>
+struct PrimitiveTypeConvertor {
+    using CppType = typename PrimitiveTypeTraits<PT>::CppType;
+    using StorageFieldType = typename PrimitiveTypeTraits<PT>::StorageFieldType;
+
+    static inline StorageFieldType&& to_storage_field_type(CppType&& value) {
+        return static_cast<StorageFieldType&&>(std::forward<CppType>(value));
+    }
+
+    static inline const StorageFieldType& to_storage_field_type(const CppType& value) {
+        return *reinterpret_cast<const StorageFieldType*>(&value);
+    }
+};
+
+template <>
+struct PrimitiveTypeConvertor<TYPE_DATE> {
+    using CppType = typename PrimitiveTypeTraits<TYPE_DATE>::CppType;
+    using StorageFieldType = typename PrimitiveTypeTraits<TYPE_DATE>::StorageFieldType;
+
+    static inline StorageFieldType to_storage_field_type(const CppType& value) {
+        return value.to_olap_date();
+    }
+};
+
+template <>
+struct PrimitiveTypeConvertor<TYPE_DATETIME> {
+    using CppType = typename PrimitiveTypeTraits<TYPE_DATETIME>::CppType;
+    using StorageFieldType = typename PrimitiveTypeTraits<TYPE_DATETIME>::StorageFieldType;
+
+    static inline StorageFieldType to_storage_field_type(const CppType& value) {
+        return value.to_olap_datetime();
+    }
+};
+
+template <>
+struct PrimitiveTypeConvertor<TYPE_DECIMALV2> {
+    using CppType = typename PrimitiveTypeTraits<TYPE_DECIMALV2>::CppType;
+    using StorageFieldType = typename PrimitiveTypeTraits<TYPE_DECIMALV2>::StorageFieldType;
+
+    static inline StorageFieldType to_storage_field_type(const CppType& value) {
+        return {value.int_value(), value.frac_value()};
+    }
+};
+
 } // namespace doris
diff --git a/be/src/util/date_func.cpp b/be/src/util/date_func.cpp
index a9198a249ab65c..d3ad3e0a1be111 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 405132c3f01282..dbf8a738e7819a 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/column_vector.h b/be/src/vec/columns/column_vector.h
index ffb91494c26c10..00a49835c6bd15 100644
--- a/be/src/vec/columns/column_vector.h
+++ b/be/src/vec/columns/column_vector.h
@@ -197,10 +197,10 @@ class ColumnVector final : public COWHelper<ColumnVectorHelper, ColumnVector<T>>
 
     void insert_date_column(const char* data_ptr, size_t num) {
         data.reserve(data.size() + num);
-        size_t input_value_size = sizeof(uint24_t);
+        constexpr size_t input_value_size = sizeof(uint24_t);
 
         for (int i = 0; i < num; i++) {
-            uint64_t val = 0;
+            uint24_t val = 0;
             memcpy((char*)(&val), data_ptr, input_value_size);
             data_ptr += input_value_size;
 
diff --git a/be/src/vec/columns/predicate_column.h b/be/src/vec/columns/predicate_column.h
index 87d1993e2747e2..1448c3d4bd08ca 100644
--- a/be/src/vec/columns/predicate_column.h
+++ b/be/src/vec/columns/predicate_column.h
@@ -41,36 +41,13 @@ namespace doris::vectorized {
 template <PrimitiveType Type>
 class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<Type>> {
 private:
-    PredicateColumnType() {}
+    PredicateColumnType() = default;
     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();
-
-        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)));
-        }
-    }
-
-    void insert_datetime_to_res_column(const uint16_t* sel, size_t sel_size,
-                                       ColumnVector<Int64>* res_ptr) {
-        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);
-        }
-    }
-
     void insert_string_to_res_column(const uint16_t* sel, size_t sel_size, ColumnString* res_ptr) {
         StringRef refs[sel_size];
         size_t length = 0;
@@ -86,26 +63,22 @@ 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) {
-        static_assert(std::is_same_v<T, Y>);
+        static_assert(std::is_same_v<ColumnContainer<Y>, ColumnType>);
         auto& res_data = res_ptr->get_data();
         DCHECK(res_data.empty());
         res_data.reserve(sel_size);
         Y* y = (Y*)res_data.get_end_ptr();
         for (size_t i = 0; i < sel_size; i++) {
-            y[i] = T(data[sel[i]]);
+            if constexpr (std::is_same_v<Y, T>) {
+                y[i] = data[sel[i]];
+            } else {
+                static_assert(sizeof(Y) == sizeof(T));
+                memcpy(reinterpret_cast<void*>(&y[i]), reinterpret_cast<void*>(&data[sel[i]]),
+                       sizeof(Y));
+            }
         }
         res_data.set_end_ptr(y + sel_size);
     }
@@ -121,13 +94,7 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
     void insert_many_default_type(const char* data_ptr, size_t num) {
         auto old_size = data.size();
         data.resize(old_size + num);
-        memcpy(data.data() + old_size, data_ptr, num * sizeof(T));
-    }
-
-    void insert_many_in_copy_way(const char* data_ptr, size_t num) {
-        auto old_size = data.size();
-        data.resize(old_size + num);
-        memcpy(data.data() + old_size, data_ptr, num * sizeof(T));
+        memcpy(reinterpret_cast<void*>(data.data() + old_size), data_ptr, num * sizeof(T));
     }
 
 public:
@@ -186,13 +153,6 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
         data.push_back_without_reserve(sv);
     }
 
-    void insert_decimal_value(const char* data_ptr, size_t length) {
-        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);
-    }
-
     // used for int128
     void insert_in_copy_way(const char* data_ptr, size_t length) {
         T val {};
@@ -208,8 +168,6 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
     void insert_data(const char* data_ptr, size_t length) override {
         if constexpr (std::is_same_v<T, StringRef>) {
             insert_string_value(data_ptr, length);
-        } else if constexpr (std::is_same_v<T, decimal12_t>) {
-            insert_decimal_value(data_ptr, length);
         } else if constexpr (std::is_same_v<T, Int128>) {
             insert_in_copy_way(data_ptr, length);
         } else {
@@ -218,29 +176,56 @@ 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);
-        char* input_data_ptr = const_cast<char*>(data_ptr);
+        constexpr size_t input_type_size = sizeof(PrimitiveTypeTraits<TYPE_DATE>::StorageFieldType);
+        static_assert(input_type_size == sizeof(uint24_t));
+        const auto* input_data_ptr = reinterpret_cast<const uint24_t*>(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;
+            res_ptr[i].set_olap_date(unaligned_load<uint24_t>(&input_data_ptr[i]));
+        }
+        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));
+        const auto* input_data_ptr = reinterpret_cast<const uint64_t*>(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(input_data_ptr[i]);
+        }
+        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);
+
+        auto* target = (DecimalV2Value*)(data.data() + old_size);
+        for (int i = 0; i < num; i++) {
+            const char* cur_ptr = data_ptr + sizeof(decimal12_t) * i;
+            auto 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);
-        } else if constexpr (std::is_same_v<T, Int128>) {
-            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, 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 actual int64.
             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);
         }
@@ -444,39 +429,12 @@ class PredicateColumnType final : public COWHelper<IColumn, PredicateColumnType<
 
     Status filter_by_selector(const uint16_t* sel, size_t sel_size, IColumn* col_ptr) override {
         ColumnType* column = assert_cast<ColumnType*>(col_ptr);
-        // 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";
-            }
-        } else if constexpr (std::is_same_v<ColumnVector<T>, ColumnType>) {
-            // DateV2 and DateTimeV2, its storage format is equal to compute format
-            // not need convert
-            insert_default_value_res_column(sel, sel_size, column);
-        } else if constexpr (std::is_same_v<ColumnDecimal<T>, ColumnType>) {
-            insert_default_value_res_column(sel, sel_size, column);
-        } else if constexpr (std::is_same_v<T, StringRef>) {
+        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>) {
+        } else if constexpr (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>) {
-            insert_default_value_res_column(
-                    sel, sel_size,
-                    reinterpret_cast<vectorized::ColumnVector<doris::vectorized::IPv4>*>(col_ptr));
-        } else if constexpr (std::is_same_v<T, doris::vectorized::IPv6>) {
-            insert_default_value_res_column(
-                    sel, sel_size,
-                    reinterpret_cast<vectorized::ColumnVector<doris::vectorized::IPv6>*>(col_ptr));
         } else {
-            return Status::NotSupported("not supported output type in predicate_column, type={}",
-                                        type_to_string(Type));
+            insert_default_value_res_column(sel, sel_size, column);
         }
         return Status::OK();
     }
diff --git a/be/src/vec/core/types.h b/be/src/vec/core/types.h
index 9bf289787d26d6..9f3b64a9d1fcf0 100644
--- a/be/src/vec/core/types.h
+++ b/be/src/vec/core/types.h
@@ -226,6 +226,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 c29dabfe62ec84..4e46accb5e07db 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/date/test_date_key.out b/regression-test/data/datatype_p0/date/test_date_key.out
new file mode 100644
index 00000000000000..2b9b9d1589e097
--- /dev/null
+++ b/regression-test/data/datatype_p0/date/test_date_key.out
@@ -0,0 +1,193 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql1 --
+2016-11-04	1
+2016-11-04	1
+
+-- !sql2 --
+2016-11-04	1
+2016-11-04	1
+2016-12-06	3
+2016-12-06	3
+2016-12-07	4
+2016-12-07	44
+2017-02-05	5
+2017-02-05	5
+2017-08-22	6
+2017-08-22	666
+
+-- !sql3 --
+2016-11-04	1
+2016-11-04	1
+2016-11-05	2
+2016-11-05	22
+
+-- !sql_in --
+2016-11-04	1
+2016-11-04	1
+2016-11-05	2
+2016-11-05	22
+2017-02-05	5
+2017-02-05	5
+2017-08-22	6
+2017-08-22	666
+
+-- !sql_not_in --
+2016-12-06	3
+2016-12-06	3
+2016-12-07	4
+2016-12-07	44
+
+-- !sql_distribute_1 --
+1	2016-11-04
+1	2016-11-04
+
+-- !sql_distribute_2 --
+1	2016-11-04
+1	2016-11-04
+3	2016-12-06
+3	2016-12-06
+4	2016-12-07
+5	2017-02-05
+5	2017-02-05
+6	2017-08-22
+6	2017-08-22
+44	2016-12-07
+
+-- !sql_distribute_3 --
+1	2016-11-04
+1	2016-11-04
+2	2016-11-05
+22	2016-11-05
+
+-- !sql_distribute_in --
+1	2016-11-04
+1	2016-11-04
+2	2016-11-05
+5	2017-02-05
+5	2017-02-05
+6	2017-08-22
+6	2017-08-22
+22	2016-11-05
+
+-- !sql_distribute_not_in --
+3	2016-12-06
+3	2016-12-06
+4	2016-12-07
+44	2016-12-07
+
+-- !sql_partition_1 --
+1	2016-11-04
+1	2016-11-04
+
+-- !sql_partition_2 --
+1	2016-11-04
+1	2016-11-04
+3	2016-12-06
+3	2016-12-06
+4	2016-12-07
+5	2017-02-05
+5	2017-02-05
+6	2017-08-22
+6	2017-08-22
+44	2016-12-07
+
+-- !sql_partition_3 --
+1	2016-11-04
+1	2016-11-04
+2	2016-11-05
+22	2016-11-05
+
+-- !sql_partition_in --
+1	2016-11-04
+1	2016-11-04
+2	2016-11-05
+5	2017-02-05
+5	2017-02-05
+6	2017-08-22
+6	2017-08-22
+22	2016-11-05
+
+-- !sql_partition_not_in --
+3	2016-12-06
+3	2016-12-06
+4	2016-12-07
+44	2016-12-07
+
+-- !join_rf --
+2016-11-05	2	2	2016-11-05
+2016-11-05	2	22	2016-11-05
+2016-11-05	22	2	2016-11-05
+2016-11-05	22	22	2016-11-05
+2016-12-07	4	4	2016-12-07
+2016-12-07	4	44	2016-12-07
+2016-12-07	44	4	2016-12-07
+2016-12-07	44	44	2016-12-07
+2017-08-22	6	6	2017-08-22
+2017-08-22	6	6	2017-08-22
+2017-08-22	666	6	2017-08-22
+2017-08-22	666	6	2017-08-22
+
+-- !join_rf2 --
+2016-11-05	2	2	2016-11-05
+2016-11-05	2	22	2016-11-05
+2016-11-05	22	2	2016-11-05
+2016-11-05	22	22	2016-11-05
+2016-12-07	4	4	2016-12-07
+2016-12-07	4	44	2016-12-07
+2016-12-07	44	4	2016-12-07
+2016-12-07	44	44	2016-12-07
+2017-08-22	6	6	2017-08-22
+2017-08-22	6	6	2017-08-22
+2017-08-22	666	6	2017-08-22
+2017-08-22	666	6	2017-08-22
+
+-- !join_rf3 --
+2016-11-05	2	2	2016-11-05
+2016-11-05	2	22	2016-11-05
+2016-11-05	22	2	2016-11-05
+2016-11-05	22	22	2016-11-05
+2016-12-07	4	4	2016-12-07
+2016-12-07	4	44	2016-12-07
+2016-12-07	44	4	2016-12-07
+2016-12-07	44	44	2016-12-07
+2017-08-22	6	6	2017-08-22
+2017-08-22	6	6	2017-08-22
+2017-08-22	666	6	2017-08-22
+2017-08-22	666	6	2017-08-22
+
+-- !key_after_del --
+2016-11-04	1
+2016-11-04	1
+2016-11-05	2
+2016-11-05	22
+2016-12-07	4
+2016-12-07	44
+2017-02-05	5
+2017-02-05	5
+2017-08-22	6
+2017-08-22	666
+
+-- !distributed_after_del --
+1	2016-11-04
+1	2016-11-04
+2	2016-11-05
+4	2016-12-07
+5	2017-02-05
+5	2017-02-05
+6	2017-08-22
+6	2017-08-22
+22	2016-11-05
+44	2016-12-07
+
+-- !partition_after_del --
+1	2016-11-04
+1	2016-11-04
+2	2016-11-05
+4	2016-12-07
+5	2017-02-05
+5	2017-02-05
+6	2017-08-22
+6	2017-08-22
+22	2016-11-05
+44	2016-12-07
+
diff --git a/regression-test/data/datatype_p0/date/test_datetime_key.out b/regression-test/data/datatype_p0/date/test_datetime_key.out
new file mode 100644
index 00000000000000..27bc0b42f1ef5d
--- /dev/null
+++ b/regression-test/data/datatype_p0/date/test_datetime_key.out
@@ -0,0 +1,193 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql1 --
+2016-11-04T06:21:44	1
+2016-11-04T06:21:44	1
+
+-- !sql2 --
+2016-11-04T06:21:44	1
+2016-11-04T06:21:44	1
+2016-12-06T22:31:09	3
+2016-12-06T22:31:09	3
+2016-12-07T23:59:59	4
+2016-12-07T23:59:59	44
+2017-02-05T23:59:59	5
+2017-02-05T23:59:59	5
+2017-08-22T00:00:01	6
+2017-08-22T00:00:01	666
+
+-- !sql3 --
+2016-11-04T06:21:44	1
+2016-11-04T06:21:44	1
+2016-11-05T12:41:14	2
+2016-11-05T12:41:14	22
+
+-- !sql_in --
+2016-11-04T06:21:44	1
+2016-11-04T06:21:44	1
+2016-11-05T12:41:14	2
+2016-11-05T12:41:14	22
+2017-02-05T23:59:59	5
+2017-02-05T23:59:59	5
+2017-08-22T00:00:01	6
+2017-08-22T00:00:01	666
+
+-- !sql_not_in --
+2016-12-06T22:31:09	3
+2016-12-06T22:31:09	3
+2016-12-07T23:59:59	4
+2016-12-07T23:59:59	44
+
+-- !sql_distribute_1 --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+
+-- !sql_distribute_2 --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+3	2016-12-06T22:31:09
+3	2016-12-06T22:31:09
+4	2016-12-07T23:59:59
+5	2017-02-05T23:59:59
+5	2017-02-05T23:59:59
+6	2017-08-22T00:00:01
+6	2017-08-22T00:00:01
+44	2016-12-07T23:59:59
+
+-- !sql_distribute_3 --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+2	2016-11-05T12:41:14
+22	2016-11-05T12:41:14
+
+-- !sql_distribute_in --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+2	2016-11-05T12:41:14
+5	2017-02-05T23:59:59
+5	2017-02-05T23:59:59
+6	2017-08-22T00:00:01
+6	2017-08-22T00:00:01
+22	2016-11-05T12:41:14
+
+-- !sql_distribute_not_in --
+3	2016-12-06T22:31:09
+3	2016-12-06T22:31:09
+4	2016-12-07T23:59:59
+44	2016-12-07T23:59:59
+
+-- !sql_partition_1 --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+
+-- !sql_distribute_2 --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+3	2016-12-06T22:31:09
+3	2016-12-06T22:31:09
+4	2016-12-07T23:59:59
+5	2017-02-05T23:59:59
+5	2017-02-05T23:59:59
+6	2017-08-22T00:00:01
+6	2017-08-22T00:00:01
+44	2016-12-07T23:59:59
+
+-- !sql_distribute_3 --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+2	2016-11-05T12:41:14
+22	2016-11-05T12:41:14
+
+-- !sql_distribute_in --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+2	2016-11-05T12:41:14
+5	2017-02-05T23:59:59
+5	2017-02-05T23:59:59
+6	2017-08-22T00:00:01
+6	2017-08-22T00:00:01
+22	2016-11-05T12:41:14
+
+-- !sql_distribute_not_in --
+3	2016-12-06T22:31:09
+3	2016-12-06T22:31:09
+4	2016-12-07T23:59:59
+44	2016-12-07T23:59:59
+
+-- !join_rf --
+2016-11-05T12:41:14	2	2	2016-11-05T12:41:14
+2016-11-05T12:41:14	2	22	2016-11-05T12:41:14
+2016-11-05T12:41:14	22	2	2016-11-05T12:41:14
+2016-11-05T12:41:14	22	22	2016-11-05T12:41:14
+2016-12-07T23:59:59	4	4	2016-12-07T23:59:59
+2016-12-07T23:59:59	4	44	2016-12-07T23:59:59
+2016-12-07T23:59:59	44	4	2016-12-07T23:59:59
+2016-12-07T23:59:59	44	44	2016-12-07T23:59:59
+2017-08-22T00:00:01	6	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	6	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	666	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	666	6	2017-08-22T00:00:01
+
+-- !join_rf2 --
+2016-11-05T12:41:14	2	2	2016-11-05T12:41:14
+2016-11-05T12:41:14	2	22	2016-11-05T12:41:14
+2016-11-05T12:41:14	22	2	2016-11-05T12:41:14
+2016-11-05T12:41:14	22	22	2016-11-05T12:41:14
+2016-12-07T23:59:59	4	4	2016-12-07T23:59:59
+2016-12-07T23:59:59	4	44	2016-12-07T23:59:59
+2016-12-07T23:59:59	44	4	2016-12-07T23:59:59
+2016-12-07T23:59:59	44	44	2016-12-07T23:59:59
+2017-08-22T00:00:01	6	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	6	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	666	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	666	6	2017-08-22T00:00:01
+
+-- !join_rf3 --
+2016-11-05T12:41:14	2	2	2016-11-05T12:41:14
+2016-11-05T12:41:14	2	22	2016-11-05T12:41:14
+2016-11-05T12:41:14	22	2	2016-11-05T12:41:14
+2016-11-05T12:41:14	22	22	2016-11-05T12:41:14
+2016-12-07T23:59:59	4	4	2016-12-07T23:59:59
+2016-12-07T23:59:59	4	44	2016-12-07T23:59:59
+2016-12-07T23:59:59	44	4	2016-12-07T23:59:59
+2016-12-07T23:59:59	44	44	2016-12-07T23:59:59
+2017-08-22T00:00:01	6	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	6	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	666	6	2017-08-22T00:00:01
+2017-08-22T00:00:01	666	6	2017-08-22T00:00:01
+
+-- !key_after_del --
+2016-11-04T06:21:44	1
+2016-11-04T06:21:44	1
+2016-11-05T12:41:14	2
+2016-11-05T12:41:14	22
+2016-12-07T23:59:59	4
+2016-12-07T23:59:59	44
+2017-02-05T23:59:59	5
+2017-02-05T23:59:59	5
+2017-08-22T00:00:01	6
+2017-08-22T00:00:01	666
+
+-- !distributed_after_del --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+2	2016-11-05T12:41:14
+4	2016-12-07T23:59:59
+5	2017-02-05T23:59:59
+5	2017-02-05T23:59:59
+6	2017-08-22T00:00:01
+6	2017-08-22T00:00:01
+22	2016-11-05T12:41:14
+44	2016-12-07T23:59:59
+
+-- !partition_after_del --
+1	2016-11-04T06:21:44
+1	2016-11-04T06:21:44
+2	2016-11-05T12:41:14
+4	2016-12-07T23:59:59
+5	2017-02-05T23:59:59
+5	2017-02-05T23:59:59
+6	2017-08-22T00:00:01
+6	2017-08-22T00:00:01
+22	2016-11-05T12:41:14
+44	2016-12-07T23:59:59
+
diff --git a/regression-test/data/datatype_p0/date/test_datev1.out b/regression-test/data/datatype_p0/date/test_datev1.out
index 8397a4d7e42d90..27401c431e18b3 100644
--- a/regression-test/data/datatype_p0/date/test_datev1.out
+++ b/regression-test/data/datatype_p0/date/test_datev1.out
@@ -15,37 +15,78 @@
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
 
--- !sql2 --
+-- !sql3 --
 1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
 
--- !sql2 --
+-- !sql4 --
 1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
 
--- !sql2 --
+-- !sql5 --
 1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
 
--- !sql2 --
+-- !sql6 --
 1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
 
--- !sql2 --
+-- !sql7 --
 1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
 
--- !sql2 --
+-- !sql8 --
 1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
 
--- !sql2 --
+-- !sql9 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql10 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql11 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql12 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql13 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql14 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql15 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql16 --
 1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
 2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
 3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
+-- !sql17 --
+1	2000-01-01	2000-01-01T11:11:11	1	2000-01-01	2000-01-01T11:11:11
+2	2000-02-02	2000-02-02T11:11:11	2	2000-02-02	2000-02-02T11:11:11
+3	2000-03-02	2000-03-02T11:11:11	3	2000-03-02	2000-03-02T11:11:11
+
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 cceb1c0f8587d2..a5c5eaad199399 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/date/test_date_exprs.groovy b/regression-test/suites/datatype_p0/date/test_date_exprs.groovy
index 6fb62b86d1e1de..f4106a9f9df5a6 100644
--- a/regression-test/suites/datatype_p0/date/test_date_exprs.groovy
+++ b/regression-test/suites/datatype_p0/date/test_date_exprs.groovy
@@ -20,20 +20,20 @@ suite("test_date_exprs") {
     def tbName = "test_date_exprs"
     sql "DROP TABLE IF EXISTS ${tbName}"
     sql """
-            create table ${tbName}(k1 datetimev2, k2 int) distributed by hash(k1) buckets 1 properties("replication_num" = "1");
+            create table ${tbName}(k1 datetimev1, k2 int) distributed by hash(k1) buckets 1 properties("replication_num" = "1");
         """
     sql """ insert into ${tbName} values("2016-11-04 00:00:01", 1); """
 
     qt_sql1 """ select dt
              from
              (
-             select cast(k1 as date) as dt
+             select cast(k1 as datev1) as dt
              from ${tbName}
              ) r; """
     qt_sql2 """ select dt
              from
              (
-             select cast(k1 as datev2) as dt
+             select cast(k1 as datev1) as dt
              from ${tbName}
              ) r; """
     sql "DROP TABLE ${tbName}"
diff --git a/regression-test/suites/datatype_p0/date/test_date_key.groovy b/regression-test/suites/datatype_p0/date/test_date_key.groovy
new file mode 100644
index 00000000000000..22c9a018dd393a
--- /dev/null
+++ b/regression-test/suites/datatype_p0/date/test_date_key.groovy
@@ -0,0 +1,195 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_date_key") {
+    sql "DROP TABLE IF EXISTS `test_date_key`"
+    sql """
+        create table `test_date_key` (
+            `k1` datev1, `k2` int,
+            INDEX idx_k1 (`k1`) USING BITMAP,
+        ) duplicate key(`k1`)
+        distributed by hash(k2) buckets 3
+        properties("replication_num" = "1");
+    """
+    sql """ insert into `test_date_key` values("2016-11-04", 1); """
+    sql """ insert into `test_date_key` values("2016-11-05", 2); """
+    sql """ insert into `test_date_key` values("2016-12-06", 3); """
+    sql """ insert into `test_date_key` values("2016-12-07", 4); """
+    sql """ insert into `test_date_key` values("2017-02-05", 5); """
+    sql """ insert into `test_date_key` values("2017-08-22", 6); """
+    sql """ insert into `test_date_key` values("2016-11-04", 1); """
+    sql """ insert into `test_date_key` values("2016-11-05", 22); """
+    sql """ insert into `test_date_key` values("2016-12-06", 3); """
+    sql """ insert into `test_date_key` values("2016-12-07", 44); """
+    sql """ insert into `test_date_key` values("2017-02-05", 5); """
+    sql """ insert into `test_date_key` values("2017-08-22", 666); """
+
+    qt_sql1 """
+        select * from `test_date_key` where `k1` = "2016-11-04" order by `k1`, `k2`;
+    """
+
+    qt_sql2 """
+        select * from `test_date_key` where `k1` <> "2016-11-05" order by `k1`, `k2`;
+    """
+
+    qt_sql3 """
+        select * from `test_date_key` where `k1` = "2016-11-05" or `k1` = "2016-11-04" order by `k1`, `k2`;
+    """
+
+    qt_sql_in """
+        select * from `test_date_key` where `k1` in ("2016-11-05", "2016-11-04", "2017-08-22", "2017-02-05")  order by `k1`, `k2`;
+    """
+
+    qt_sql_not_in """
+        select * from `test_date_key` where `k1` not in ("2016-11-05", "2016-11-04", "2017-08-22", "2017-02-05")  order by `k1`, `k2`;
+    """
+
+    sql "DROP TABLE IF EXISTS `test_date_distributed`"
+    sql """
+        create table `test_date_distributed` (
+            `k1` int, `k2` datev1
+        ) duplicate key(`k1`)
+        distributed by hash(k2) buckets 3
+        properties("replication_num" = "1");
+    """
+    sql """ insert into `test_date_distributed` values(1, "2016-11-04"); """
+    sql """ insert into `test_date_distributed` values(2, "2016-11-05"); """
+    sql """ insert into `test_date_distributed` values(3, "2016-12-06"); """
+    sql """ insert into `test_date_distributed` values(4, "2016-12-07"); """
+    sql """ insert into `test_date_distributed` values(5, "2017-02-05"); """
+    sql """ insert into `test_date_distributed` values(6, "2017-08-22"); """
+    sql """ insert into `test_date_distributed` values(1, "2016-11-04"); """
+    sql """ insert into `test_date_distributed` values(22, "2016-11-05");"""
+    sql """ insert into `test_date_distributed` values(3, "2016-12-06"); """
+    sql """ insert into `test_date_distributed` values(44, "2016-12-07");"""
+    sql """ insert into `test_date_distributed` values(5, "2017-02-05"); """
+    sql """ insert into `test_date_distributed` values(6, "2017-08-22"); """
+
+    qt_sql_distribute_1 """
+        select * from `test_date_distributed` where `k2` = "2016-11-04" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_2 """
+        select * from `test_date_distributed` where `k2` <> "2016-11-05" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_3 """
+        select * from `test_date_distributed` where `k2` = "2016-11-05" or `k2` = "2016-11-04" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_in """
+        select * from `test_date_distributed` where `k2` in ("2016-11-05", "2016-11-04", "2017-08-22", "2017-02-05")  order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_not_in """
+        select * from `test_date_distributed` where `k2` not in ("2016-11-05", "2016-11-04", "2017-08-22", "2017-02-05")  order by `k1`, `k2`;
+    """
+
+    sql "DROP TABLE IF EXISTS `test_date_partition`"
+    sql """
+        create table `test_date_partition` (
+            `k1` int, `k2` datev1,
+            INDEX idx_k2 (`k2`) USING BITMAP
+        ) duplicate key(`k1`)
+        PARTITION BY range(`k2`)(
+            PARTITION p_1610 VALUES [('2016-10-01'), ('2016-10-31')),
+            PARTITION p_1611 VALUES [('2016-11-01'), ('2016-11-30')),
+            PARTITION p_1612 VALUES [('2016-12-01'), ('2016-12-31')),
+            PARTITION p_1702 VALUES [('2017-02-01'), ('2017-02-28')),
+            PARTITION p_1708 VALUES [('2017-08-01'), ('2017-08-31'))
+        )
+        distributed by hash(`k1`) buckets 3
+        properties(
+            "replication_num" = "1"
+        );
+    """
+
+    sql """ insert into `test_date_partition` values(1, "2016-11-04"); """
+    sql """ insert into `test_date_partition` values(2, "2016-11-05"); """
+    sql """ insert into `test_date_partition` values(3, "2016-12-06"); """
+    sql """ insert into `test_date_partition` values(4, "2016-12-07"); """
+    sql """ insert into `test_date_partition` values(5, "2017-02-05"); """
+    sql """ insert into `test_date_partition` values(6, "2017-08-22"); """
+    sql """ insert into `test_date_partition` values(1, "2016-11-04"); """
+    sql """ insert into `test_date_partition` values(22, "2016-11-05");"""
+    sql """ insert into `test_date_partition` values(3, "2016-12-06"); """
+    sql """ insert into `test_date_partition` values(44, "2016-12-07");"""
+    sql """ insert into `test_date_partition` values(5, "2017-02-05"); """
+    sql """ insert into `test_date_partition` values(6, "2017-08-22"); """
+
+    qt_sql_partition_1 """
+        select * from `test_date_partition` where `k2` = "2016-11-04" order by `k1`, `k2`;
+    """
+
+    qt_sql_partition_2 """
+        select * from `test_date_partition` where `k2` <> "2016-11-05" order by `k1`, `k2`;
+    """
+
+    qt_sql_partition_3 """
+        select * from `test_date_partition` where `k2` = "2016-11-05" or `k2` = "2016-11-04" order by `k1`, `k2`;
+    """
+
+    qt_sql_partition_in """
+        select * from `test_date_partition` where `k2` in ("2016-11-05", "2016-11-04", "2017-08-22", "2017-02-05")  order by `k1`, `k2`;
+    """
+
+    qt_sql_partition_not_in """
+        select * from `test_date_partition` where `k2` not in ("2016-11-05", "2016-11-04", "2017-08-22", "2017-02-05")  order by `k1`, `k2`;
+    """
+
+    qt_join_rf """
+        select *
+        from `test_date_key` `t1`, `test_date_distributed` `t2`
+        where `t1`.`k1` = `t2`.`k2` and `t1`.`k2` % 2 = 0 order by `t1`.`k1`, `t1`.`k2`, `t2`.`k1`, `t2`.`k2`;
+    """
+
+    qt_join_rf2 """
+        select *
+        from `test_date_key` `t1`, `test_date_distributed` `t2`
+        where `t1`.`k1` = `t2`.`k2` and `t2`.`k1` % 2 = 0 order by `t1`.`k1`, `t1`.`k2`, `t2`.`k1`, `t2`.`k2`;
+    """
+
+    qt_join_rf3 """
+        select *
+        from `test_date_key` `t1`, `test_date_partition` `t2`
+        where `t1`.`k1` = `t2`.`k2` and `t2`.`k1` % 2 = 0 order by `t1`.`k1`, `t1`.`k2`, `t2`.`k1`, `t2`.`k2`;
+    """
+
+    sql """
+        delete from `test_date_key` where `k1` = '2016-12-06';
+    """
+
+    sql """
+        delete from `test_date_distributed` where `k2` = '2016-12-06';
+    """
+
+    sql """
+        delete from `test_date_partition` where `k2` = '2016-12-06';
+    """
+
+    qt_key_after_del """
+        select * from `test_date_key` order by `k1`, `k2`;
+    """
+
+    qt_distributed_after_del """
+        select * from `test_date_distributed` order by `k1`, `k2`;
+    """
+
+    qt_partition_after_del """
+        select * from `test_date_partition` order by `k1`, `k2`;
+    """
+}
diff --git a/regression-test/suites/datatype_p0/date/test_date_runtime_filter.groovy b/regression-test/suites/datatype_p0/date/test_date_runtime_filter.groovy
index 109c0b5e85cf64..1933193d796df2 100644
--- a/regression-test/suites/datatype_p0/date/test_date_runtime_filter.groovy
+++ b/regression-test/suites/datatype_p0/date/test_date_runtime_filter.groovy
@@ -22,8 +22,8 @@ suite("test_date_runtime_filter") {
     sql """
             CREATE TABLE IF NOT EXISTS ${tbName} (
                 c0 int,
-                c2 date,
-                c3 datetime
+                c2 datev1,
+                c3 datetimev1
             )
             DISTRIBUTED BY HASH(c0) BUCKETS 5 properties("replication_num" = "1");
         """
diff --git a/regression-test/suites/datatype_p0/date/test_datetime_key.groovy b/regression-test/suites/datatype_p0/date/test_datetime_key.groovy
new file mode 100644
index 00000000000000..848cdc1a4c58af
--- /dev/null
+++ b/regression-test/suites/datatype_p0/date/test_datetime_key.groovy
@@ -0,0 +1,195 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_datetime_key") {
+    sql "DROP TABLE IF EXISTS `test_datetime_key`"
+    sql """
+        create table `test_datetime_key` (
+            `k1` datetimev1, `k2` int,
+            INDEX idx_k1 (`k1`) USING BITMAP,
+        ) duplicate key(`k1`)
+        distributed by hash(k2) buckets 3
+        properties("replication_num" = "1");
+    """
+    sql """ insert into `test_datetime_key` values("2016-11-04 06:21:44", 1); """
+    sql """ insert into `test_datetime_key` values("2016-11-05 12:41:14", 2); """
+    sql """ insert into `test_datetime_key` values("2016-12-06 22:31:09", 3); """
+    sql """ insert into `test_datetime_key` values("2016-12-07 23:59:59", 4); """
+    sql """ insert into `test_datetime_key` values("2017-02-05 23:59:59", 5); """
+    sql """ insert into `test_datetime_key` values("2017-08-22 00:00:01", 6); """
+    sql """ insert into `test_datetime_key` values("2016-11-04 06:21:44", 1); """
+    sql """ insert into `test_datetime_key` values("2016-11-05 12:41:14", 22); """
+    sql """ insert into `test_datetime_key` values("2016-12-06 22:31:09", 3); """
+    sql """ insert into `test_datetime_key` values("2016-12-07 23:59:59", 44); """
+    sql """ insert into `test_datetime_key` values("2017-02-05 23:59:59", 5); """
+    sql """ insert into `test_datetime_key` values("2017-08-22 00:00:01", 666); """
+
+    qt_sql1 """
+        select * from `test_datetime_key` where `k1` = "2016-11-04 06:21:44" order by `k1`, `k2`;
+    """
+
+    qt_sql2 """
+        select * from `test_datetime_key` where `k1` <> "2016-11-05 12:41:14" order by `k1`, `k2`;
+    """
+
+    qt_sql3 """
+        select * from `test_datetime_key` where `k1` = "2016-11-05 12:41:14" or `k1` = "2016-11-04 06:21:44" order by `k1`, `k2`;
+    """
+
+    qt_sql_in """
+        select * from `test_datetime_key` where `k1` in ("2016-11-05 12:41:14", "2016-11-04 06:21:44", "2017-08-22 00:00:01", "2017-02-05 23:59:59")  order by `k1`, `k2`;
+    """
+
+    qt_sql_not_in """
+        select * from `test_datetime_key` where `k1` not in ("2016-11-05 12:41:14", "2016-11-04 06:21:44", "2017-08-22 00:00:01", "2017-02-05 23:59:59")  order by `k1`, `k2`;
+    """
+
+    sql "DROP TABLE IF EXISTS `test_datetime_distributed`"
+    sql """
+        create table `test_datetime_distributed` (
+            `k1` int, `k2` datetimev1
+        ) duplicate key(`k1`)
+        distributed by hash(k2) buckets 3
+        properties("replication_num" = "1");
+    """
+    sql """ insert into `test_datetime_distributed` values(1, "2016-11-04 06:21:44"); """
+    sql """ insert into `test_datetime_distributed` values(2, "2016-11-05 12:41:14"); """
+    sql """ insert into `test_datetime_distributed` values(3, "2016-12-06 22:31:09"); """
+    sql """ insert into `test_datetime_distributed` values(4, "2016-12-07 23:59:59"); """
+    sql """ insert into `test_datetime_distributed` values(5, "2017-02-05 23:59:59"); """
+    sql """ insert into `test_datetime_distributed` values(6, "2017-08-22 00:00:01"); """
+    sql """ insert into `test_datetime_distributed` values(1, "2016-11-04 06:21:44"); """
+    sql """ insert into `test_datetime_distributed` values(22, "2016-11-05 12:41:14");"""
+    sql """ insert into `test_datetime_distributed` values(3, "2016-12-06 22:31:09"); """
+    sql """ insert into `test_datetime_distributed` values(44, "2016-12-07 23:59:59");"""
+    sql """ insert into `test_datetime_distributed` values(5, "2017-02-05 23:59:59"); """
+    sql """ insert into `test_datetime_distributed` values(6, "2017-08-22 00:00:01"); """
+
+    qt_sql_distribute_1 """
+        select * from `test_datetime_distributed` where `k2` = "2016-11-04 06:21:44" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_2 """
+        select * from `test_datetime_distributed` where `k2` <> "2016-11-05 12:41:14" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_3 """
+        select * from `test_datetime_distributed` where `k2` = "2016-11-05 12:41:14" or `k2` = "2016-11-04 06:21:44" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_in """
+        select * from `test_datetime_distributed` where `k2` in ("2016-11-05 12:41:14", "2016-11-04 06:21:44", "2017-08-22 00:00:01", "2017-02-05 23:59:59")  order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_not_in """
+        select * from `test_datetime_distributed` where `k2` not in ("2016-11-05 12:41:14", "2016-11-04 06:21:44", "2017-08-22 00:00:01", "2017-02-05 23:59:59")  order by `k1`, `k2`;
+    """
+
+    sql "DROP TABLE IF EXISTS `test_datetime_partition`"
+    sql """
+        create table `test_datetime_partition` (
+            `k1` int, `k2` datetimev1,
+            INDEX idx_k2 (`k2`) USING BITMAP
+        ) duplicate key(`k1`)
+        PARTITION BY range(`k2`)(
+            PARTITION p_1610 VALUES [('2016-10-01 00:00:00'), ('2016-10-31 23:59:59')),
+            PARTITION p_1611 VALUES [('2016-11-01 00:00:00'), ('2016-11-30 23:59:59')),
+            PARTITION p_1612 VALUES [('2016-12-01 00:00:00'), ('2016-12-31 23:59:59')),
+            PARTITION p_1702 VALUES [('2017-02-01 00:00:00'), ('2017-02-28 23:59:59')),
+            PARTITION p_1708 VALUES [('2017-08-01 00:00:00'), ('2017-08-31 23:59:59'))
+        )
+        distributed by hash(`k1`) buckets 3
+        properties(
+            "replication_num" = "1"
+        );
+    """
+
+    sql """ insert into `test_datetime_partition` values(1, "2016-11-04 06:21:44"); """
+    sql """ insert into `test_datetime_partition` values(2, "2016-11-05 12:41:14"); """
+    sql """ insert into `test_datetime_partition` values(3, "2016-12-06 22:31:09"); """
+    sql """ insert into `test_datetime_partition` values(4, "2016-12-07 23:59:59"); """
+    sql """ insert into `test_datetime_partition` values(5, "2017-02-05 23:59:59"); """
+    sql """ insert into `test_datetime_partition` values(6, "2017-08-22 00:00:01"); """
+    sql """ insert into `test_datetime_partition` values(1, "2016-11-04 06:21:44"); """
+    sql """ insert into `test_datetime_partition` values(22, "2016-11-05 12:41:14");"""
+    sql """ insert into `test_datetime_partition` values(3, "2016-12-06 22:31:09"); """
+    sql """ insert into `test_datetime_partition` values(44, "2016-12-07 23:59:59");"""
+    sql """ insert into `test_datetime_partition` values(5, "2017-02-05 23:59:59"); """
+    sql """ insert into `test_datetime_partition` values(6, "2017-08-22 00:00:01"); """
+
+    qt_sql_partition_1 """
+        select * from `test_datetime_partition` where `k2` = "2016-11-04 06:21:44" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_2 """
+        select * from `test_datetime_partition` where `k2` <> "2016-11-05 12:41:14" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_3 """
+        select * from `test_datetime_partition` where `k2` = "2016-11-05 12:41:14" or `k2` = "2016-11-04 06:21:44" order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_in """
+        select * from `test_datetime_partition` where `k2` in ("2016-11-05 12:41:14", "2016-11-04 06:21:44", "2017-08-22 00:00:01", "2017-02-05 23:59:59")  order by `k1`, `k2`;
+    """
+
+    qt_sql_distribute_not_in """
+        select * from `test_datetime_partition` where `k2` not in ("2016-11-05 12:41:14", "2016-11-04 06:21:44", "2017-08-22 00:00:01", "2017-02-05 23:59:59")  order by `k1`, `k2`;
+    """
+
+    qt_join_rf """
+        select *
+        from `test_datetime_key` `t1`, `test_datetime_distributed` `t2`
+        where `t1`.`k1` = `t2`.`k2` and `t1`.`k2` % 2 = 0 order by `t1`.`k1`, `t1`.`k2`, `t2`.`k1`, `t2`.`k2`;
+    """
+
+    qt_join_rf2 """
+        select *
+        from `test_datetime_key` `t1`, `test_datetime_distributed` `t2`
+        where `t1`.`k1` = `t2`.`k2` and `t2`.`k1` % 2 = 0 order by `t1`.`k1`, `t1`.`k2`, `t2`.`k1`, `t2`.`k2`;
+    """
+
+    qt_join_rf3 """
+        select *
+        from `test_datetime_key` `t1`, `test_datetime_partition` `t2`
+        where `t1`.`k1` = `t2`.`k2` and `t2`.`k1` % 2 = 0 order by `t1`.`k1`, `t1`.`k2`, `t2`.`k1`, `t2`.`k2`;
+    """
+
+    sql """
+        delete from `test_datetime_key` where `k1` = '2016-12-06 22:31:09';
+    """
+
+    sql """
+        delete from `test_datetime_distributed` where `k2` = '2016-12-06 22:31:09';
+    """
+
+    sql """
+        delete from `test_datetime_partition` where `k2` = '2016-12-06 22:31:09';
+    """
+
+    qt_key_after_del """
+        select * from `test_datetime_key` order by `k1`, `k2`;
+    """
+
+    qt_distributed_after_del """
+        select * from `test_datetime_distributed` order by `k1`, `k2`;
+    """
+
+    qt_partition_after_del """
+        select * from `test_datetime_partition` order by `k1`, `k2`;
+    """
+}
diff --git a/regression-test/suites/datatype_p0/date/test_datev1.groovy b/regression-test/suites/datatype_p0/date/test_datev1.groovy
index f99c2fb0d07f2e..6b73dfac166185 100644
--- a/regression-test/suites/datatype_p0/date/test_datev1.groovy
+++ b/regression-test/suites/datatype_p0/date/test_datev1.groovy
@@ -33,7 +33,7 @@ suite("test_datev1") {
     qt_sql2 """ select dt
              from
              (
-             select cast(k1 as date) as dt
+             select cast(k1 as datev1) as dt
              from ${tbName}
              ) r; """
     sql "DROP TABLE ${tbName}"
@@ -56,29 +56,37 @@ suite("test_datev1") {
 
     sql " set runtime_filter_type = 1; "
     qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql3 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
     sql " set runtime_filter_type = 2; "
-    qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql4 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql5 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
     sql " set runtime_filter_type = 4; "
-    qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql6 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql7 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
     sql " set runtime_filter_type = 8; "
-    qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql8 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql9 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
     sql " set runtime_filter_wait_time_ms = 0; "
 
     sql " set runtime_filter_type = 1; "
-    qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql10 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql11 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
     sql " set runtime_filter_type = 2; "
-    qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql12 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql13 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
     sql " set runtime_filter_type = 4; "
-    qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql14 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql15 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
     sql " set runtime_filter_type = 8; "
-    qt_sql2 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql16 "select * from ${tbName} a, ${tbName} b WHERE a.c3 = b.c3 ORDER BY a.c2"
+    qt_sql17 "select * from ${tbName} a, ${tbName} b WHERE a.c2 = b.c2 ORDER BY a.c2"
 
-    sql "DROP TABLE ${tbName}"
+//     sql "DROP TABLE ${tbName}"
 }
diff --git a/regression-test/suites/datatype_p0/date/test_invalid_date.groovy b/regression-test/suites/datatype_p0/date/test_invalid_date.groovy
index 6b683d90b62763..5a7af4703577db 100644
--- a/regression-test/suites/datatype_p0/date/test_invalid_date.groovy
+++ b/regression-test/suites/datatype_p0/date/test_invalid_date.groovy
@@ -23,7 +23,7 @@ suite("test_invalid_date") {
             CREATE TABLE IF NOT EXISTS ${tbName} (
                 c0 int,
                 c1 char(10),
-                c2 date,
+                c2 datev1,
                 c3 datev2
             )
             UNIQUE KEY(c0)
diff --git a/regression-test/suites/datatype_p0/nested_types/negative_cases/test_nested_types_insert_into_with_dup_table.groovy b/regression-test/suites/datatype_p0/nested_types/negative_cases/test_nested_types_insert_into_with_dup_table.groovy
index f6ce940868db5f..586c83adb7f460 100644
--- a/regression-test/suites/datatype_p0/nested_types/negative_cases/test_nested_types_insert_into_with_dup_table.groovy
+++ b/regression-test/suites/datatype_p0/nested_types/negative_cases/test_nested_types_insert_into_with_dup_table.groovy
@@ -20,6 +20,26 @@ import org.codehaus.groovy.runtime.IOGroovyMethods
 suite("test_nested_types_insert_into_with_dup_table", "p0") {
     sql 'use regression_test_datatype_p0_nested_types'
     sql 'set enable_nereids_planner=false'
+
+    sql """
+        truncate table `tbl_array_nested_types_dup`;
+    """
+
+    sql """
+        truncate table `tbl_array_nested_types_dup2`;
+    """
+
+    sql """
+        truncate table `tbl_map_types_dup`;
+    """
+
+    sql """
+        truncate table `tbl_array_map_types_dup`;
+    """
+
+    sql """
+        truncate table `tbl_map_array_types_dup`;
+    """
     
     // test action for scala to array with scala type
     //  current we support char family to insert nested type
diff --git a/regression-test/suites/datatype_p0/scalar_types/load.groovy b/regression-test/suites/datatype_p0/scalar_types/load.groovy
index 7577bbb819ce8e..f360097b98bfa9 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,