From a044b5dcc5bf9cb825f32ecc1dfdbb2c09f2e8c6 Mon Sep 17 00:00:00 2001
From: Gabriel <gabrielleebuaa@gmail.com>
Date: Mon, 11 Jul 2022 09:21:01 +0800
Subject: [PATCH] [refactor](predicate) refactor predicates in scan node
 (#10701)

* [reafactor](predicate) refactor predicates in scan node

* update
---
 be/src/exec/olap_common.cpp         |  21 ++-
 be/src/exec/olap_common.h           | 199 +++++++++++++++-------------
 be/src/exec/olap_scan_node.cpp      | 130 ++++++++++--------
 be/src/exec/olap_scan_node.h        |  16 +--
 be/src/runtime/primitive_type.h     |   6 +
 be/src/vec/exec/volap_scan_node.cpp | 131 +++++++++---------
 be/src/vec/exec/volap_scan_node.h   |  16 +--
 be/test/exec/olap_common_test.cpp   |  60 ++++-----
 8 files changed, 322 insertions(+), 257 deletions(-)

diff --git a/be/src/exec/olap_common.cpp b/be/src/exec/olap_common.cpp
index 6b0ac960fa..4760b5a151 100644
--- a/be/src/exec/olap_common.cpp
+++ b/be/src/exec/olap_common.cpp
@@ -42,17 +42,32 @@ std::string cast_to_string(int8_t value) {
 }
 
 template <>
-void ColumnValueRange<StringValue>::convert_to_fixed_value() {
+void ColumnValueRange<PrimitiveType::TYPE_STRING>::convert_to_fixed_value() {
     return;
 }
 
 template <>
-void ColumnValueRange<DecimalV2Value>::convert_to_fixed_value() {
+void ColumnValueRange<PrimitiveType::TYPE_CHAR>::convert_to_fixed_value() {
     return;
 }
 
 template <>
-void ColumnValueRange<__int128>::convert_to_fixed_value() {
+void ColumnValueRange<PrimitiveType::TYPE_VARCHAR>::convert_to_fixed_value() {
+    return;
+}
+
+template <>
+void ColumnValueRange<PrimitiveType::TYPE_HLL>::convert_to_fixed_value() {
+    return;
+}
+
+template <>
+void ColumnValueRange<PrimitiveType::TYPE_DECIMALV2>::convert_to_fixed_value() {
+    return;
+}
+
+template <>
+void ColumnValueRange<PrimitiveType::TYPE_LARGEINT>::convert_to_fixed_value() {
     return;
 }
 
diff --git a/be/src/exec/olap_common.h b/be/src/exec/olap_common.h
index 952e092cab..0484eb93c0 100644
--- a/be/src/exec/olap_common.h
+++ b/be/src/exec/olap_common.h
@@ -27,6 +27,7 @@
 
 #include "exec/olap_utils.h"
 #include "olap/tuple.h"
+#include "runtime/primitive_type.h"
 #include "runtime/type_limit.h"
 
 namespace doris {
@@ -44,25 +45,26 @@ std::string cast_to_string(int8_t);
 /**
  * @brief Column's value range
  **/
-template <class T>
+template <PrimitiveType primitive_type>
 class ColumnValueRange {
 public:
-    typedef typename std::set<T>::iterator iterator_type;
+    using CppType = typename PrimitiveTypeTraits<primitive_type>::CppType;
+    using IteratorType = typename std::set<CppType>::iterator;
 
     ColumnValueRange();
 
-    ColumnValueRange(std::string col_name, PrimitiveType type);
+    ColumnValueRange(std::string col_name);
 
-    ColumnValueRange(std::string col_name, PrimitiveType type, const T& min, const T& max,
+    ColumnValueRange(std::string col_name, const CppType& min, const CppType& max,
                      bool contain_null);
 
     // should add fixed value before add range
-    Status add_fixed_value(const T& value);
+    Status add_fixed_value(const CppType& value);
 
     // should remove fixed value after add fixed value
-    void remove_fixed_value(const T& value);
+    void remove_fixed_value(const CppType& value);
 
-    Status add_range(SQLFilterOp op, T value);
+    Status add_range(SQLFilterOp op, CppType value);
 
     bool is_fixed_value_range() const;
 
@@ -80,9 +82,9 @@ public:
 
     void convert_to_range_value();
 
-    bool has_intersection(ColumnValueRange<T>& range);
+    bool has_intersection(ColumnValueRange<primitive_type>& range);
 
-    void intersection(ColumnValueRange<T>& range);
+    void intersection(ColumnValueRange<primitive_type>& range);
 
     void set_empty_value_range() {
         _fixed_values.clear();
@@ -91,11 +93,11 @@ public:
         _contain_null = false;
     }
 
-    const std::set<T>& get_fixed_value_set() const { return _fixed_values; }
+    const std::set<CppType>& get_fixed_value_set() const { return _fixed_values; }
 
-    T get_range_max_value() const { return _high_value; }
+    CppType get_range_max_value() const { return _high_value; }
 
-    T get_range_min_value() const { return _low_value; }
+    CppType get_range_min_value() const { return _low_value; }
 
     bool is_low_value_mininum() const { return _low_value == TYPE_MIN; }
 
@@ -212,37 +214,37 @@ public:
         _contain_null = contain_null;
     };
 
-    static void add_fixed_value_range(ColumnValueRange<T>& range, T* value) {
+    static void add_fixed_value_range(ColumnValueRange<primitive_type>& range, CppType* value) {
         range.add_fixed_value(*value);
     }
 
-    static void remove_fixed_value_range(ColumnValueRange<T>& range, T* value) {
+    static void remove_fixed_value_range(ColumnValueRange<primitive_type>& range, CppType* value) {
         range.remove_fixed_value(*value);
     }
 
-    static ColumnValueRange<T> create_empty_column_value_range(PrimitiveType type) {
-        return ColumnValueRange<T>::create_empty_column_value_range("", type);
+    static ColumnValueRange<primitive_type> create_empty_column_value_range() {
+        return ColumnValueRange<primitive_type>::create_empty_column_value_range("");
     }
 
-    static ColumnValueRange<T> create_empty_column_value_range(const std::string& col_name,
-                                                               PrimitiveType type) {
-        return ColumnValueRange<T>(col_name, type, TYPE_MAX, TYPE_MIN, false);
+    static ColumnValueRange<primitive_type> create_empty_column_value_range(
+            const std::string& col_name) {
+        return ColumnValueRange<primitive_type>(col_name, TYPE_MAX, TYPE_MIN, false);
     }
 
 protected:
-    bool is_in_range(const T& value);
+    bool is_in_range(const CppType& value);
 
 private:
-    const static T TYPE_MIN; // Column type's min value
-    const static T TYPE_MAX; // Column type's max value
+    const static CppType TYPE_MIN; // Column type's min value
+    const static CppType TYPE_MAX; // Column type's max value
 
     std::string _column_name;
     PrimitiveType _column_type; // Column type (eg: TINYINT,SMALLINT,INT,BIGINT)
-    T _low_value;               // Column's low value, closed interval at left
-    T _high_value;              // Column's high value, open interval at right
+    CppType _low_value;         // Column's low value, closed interval at left
+    CppType _high_value;        // Column's high value, open interval at right
     SQLFilterOp _low_op;
     SQLFilterOp _high_op;
-    std::set<T> _fixed_values; // Column's fixed int value
+    std::set<CppType> _fixed_values; // Column's fixed int value
 
     bool _contain_null;
 };
@@ -255,8 +257,9 @@ public:
               _end_include(true),
               _is_convertible(true) {}
 
-    template <class T>
-    Status extend_scan_key(ColumnValueRange<T>& range, int32_t max_scan_key_num, bool* exact_value);
+    template <PrimitiveType primitive_type>
+    Status extend_scan_key(ColumnValueRange<primitive_type>& range, int32_t max_scan_key_num,
+                           bool* exact_value);
 
     Status get_key_range(std::vector<std::unique_ptr<OlapScanRange>>* key_range);
 
@@ -311,38 +314,44 @@ private:
     bool _is_convertible;
 };
 
-typedef std::variant<ColumnValueRange<int8_t>, ColumnValueRange<int16_t>, ColumnValueRange<int32_t>,
-                     ColumnValueRange<int64_t>, ColumnValueRange<__int128>,
-                     ColumnValueRange<StringValue>, ColumnValueRange<DateTimeValue>,
-                     ColumnValueRange<DecimalV2Value>, ColumnValueRange<bool>,
-                     ColumnValueRange<doris::vectorized::DateV2Value>>
+typedef std::variant<ColumnValueRange<TYPE_TINYINT>, ColumnValueRange<TYPE_SMALLINT>,
+                     ColumnValueRange<TYPE_INT>, ColumnValueRange<TYPE_BIGINT>,
+                     ColumnValueRange<TYPE_LARGEINT>, ColumnValueRange<TYPE_CHAR>,
+                     ColumnValueRange<TYPE_VARCHAR>, ColumnValueRange<TYPE_STRING>,
+                     ColumnValueRange<TYPE_DATE>, ColumnValueRange<TYPE_DATEV2>,
+                     ColumnValueRange<TYPE_DATETIME>, ColumnValueRange<TYPE_DECIMALV2>,
+                     ColumnValueRange<TYPE_BOOLEAN>, ColumnValueRange<TYPE_HLL>>
         ColumnValueRangeType;
 
-template <class T>
-const T ColumnValueRange<T>::TYPE_MIN = type_limit<T>::min();
-template <class T>
-const T ColumnValueRange<T>::TYPE_MAX = type_limit<T>::max();
+template <PrimitiveType primitive_type>
+const typename ColumnValueRange<primitive_type>::CppType
+        ColumnValueRange<primitive_type>::TYPE_MIN =
+                type_limit<typename ColumnValueRange<primitive_type>::CppType>::min();
+template <PrimitiveType primitive_type>
+const typename ColumnValueRange<primitive_type>::CppType
+        ColumnValueRange<primitive_type>::TYPE_MAX =
+                type_limit<typename ColumnValueRange<primitive_type>::CppType>::max();
 
-template <class T>
-ColumnValueRange<T>::ColumnValueRange() : _column_type(INVALID_TYPE) {}
+template <PrimitiveType primitive_type>
+ColumnValueRange<primitive_type>::ColumnValueRange() : _column_type(INVALID_TYPE) {}
 
-template <class T>
-ColumnValueRange<T>::ColumnValueRange(std::string col_name, PrimitiveType type)
-        : ColumnValueRange(std::move(col_name), type, TYPE_MIN, TYPE_MAX, true) {}
+template <PrimitiveType primitive_type>
+ColumnValueRange<primitive_type>::ColumnValueRange(std::string col_name)
+        : ColumnValueRange(std::move(col_name), TYPE_MIN, TYPE_MAX, true) {}
 
-template <class T>
-ColumnValueRange<T>::ColumnValueRange(std::string col_name, PrimitiveType type, const T& min,
-                                      const T& max, bool contain_null)
+template <PrimitiveType primitive_type>
+ColumnValueRange<primitive_type>::ColumnValueRange(std::string col_name, const CppType& min,
+                                                   const CppType& max, bool contain_null)
         : _column_name(std::move(col_name)),
-          _column_type(type),
+          _column_type(primitive_type),
           _low_value(min),
           _high_value(max),
           _low_op(FILTER_LARGER_OR_EQUAL),
           _high_op(FILTER_LESS_OR_EQUAL),
           _contain_null(contain_null) {}
 
-template <class T>
-Status ColumnValueRange<T>::add_fixed_value(const T& value) {
+template <PrimitiveType primitive_type>
+Status ColumnValueRange<primitive_type>::add_fixed_value(const CppType& value) {
     if (INVALID_TYPE == _column_type) {
         return Status::InternalError("AddFixedValue failed, Invalid type");
     }
@@ -356,23 +365,23 @@ Status ColumnValueRange<T>::add_fixed_value(const T& value) {
     return Status::OK();
 }
 
-template <class T>
-void ColumnValueRange<T>::remove_fixed_value(const T& value) {
+template <PrimitiveType primitive_type>
+void ColumnValueRange<primitive_type>::remove_fixed_value(const CppType& value) {
     _fixed_values.erase(value);
 }
 
-template <class T>
-bool ColumnValueRange<T>::is_fixed_value_range() const {
+template <PrimitiveType primitive_type>
+bool ColumnValueRange<primitive_type>::is_fixed_value_range() const {
     return _fixed_values.size() != 0;
 }
 
-template <class T>
-bool ColumnValueRange<T>::is_scope_value_range() const {
+template <PrimitiveType primitive_type>
+bool ColumnValueRange<primitive_type>::is_scope_value_range() const {
     return _high_value > _low_value;
 }
 
-template <class T>
-bool ColumnValueRange<T>::is_empty_value_range() const {
+template <PrimitiveType primitive_type>
+bool ColumnValueRange<primitive_type>::is_empty_value_range() const {
     if (INVALID_TYPE == _column_type) {
         return true;
     }
@@ -380,8 +389,8 @@ bool ColumnValueRange<T>::is_empty_value_range() const {
     return !is_fixed_value_range() && !is_scope_value_range() && !contain_null();
 }
 
-template <class T>
-bool ColumnValueRange<T>::is_fixed_value_convertible() const {
+template <PrimitiveType primitive_type>
+bool ColumnValueRange<primitive_type>::is_fixed_value_convertible() const {
     if (is_fixed_value_range()) {
         return false;
     }
@@ -393,8 +402,8 @@ bool ColumnValueRange<T>::is_fixed_value_convertible() const {
     return true;
 }
 
-template <class T>
-bool ColumnValueRange<T>::is_range_value_convertible() const {
+template <PrimitiveType primitive_type>
+bool ColumnValueRange<primitive_type>::is_range_value_convertible() const {
     if (!is_fixed_value_range()) {
         return false;
     }
@@ -406,8 +415,8 @@ bool ColumnValueRange<T>::is_range_value_convertible() const {
     return true;
 }
 
-template <class T>
-size_t ColumnValueRange<T>::get_convertible_fixed_value_size() const {
+template <PrimitiveType primitive_type>
+size_t ColumnValueRange<primitive_type>::get_convertible_fixed_value_size() const {
     if (!is_fixed_value_convertible()) {
         return 0;
     }
@@ -416,22 +425,31 @@ size_t ColumnValueRange<T>::get_convertible_fixed_value_size() const {
 }
 
 template <>
-void ColumnValueRange<StringValue>::convert_to_fixed_value();
+void ColumnValueRange<PrimitiveType::TYPE_STRING>::convert_to_fixed_value();
 
 template <>
-void ColumnValueRange<DecimalV2Value>::convert_to_fixed_value();
+void ColumnValueRange<PrimitiveType::TYPE_CHAR>::convert_to_fixed_value();
 
 template <>
-void ColumnValueRange<__int128>::convert_to_fixed_value();
+void ColumnValueRange<PrimitiveType::TYPE_VARCHAR>::convert_to_fixed_value();
 
-template <class T>
-void ColumnValueRange<T>::convert_to_fixed_value() {
+template <>
+void ColumnValueRange<PrimitiveType::TYPE_HLL>::convert_to_fixed_value();
+
+template <>
+void ColumnValueRange<PrimitiveType::TYPE_DECIMALV2>::convert_to_fixed_value();
+
+template <>
+void ColumnValueRange<PrimitiveType::TYPE_LARGEINT>::convert_to_fixed_value();
+
+template <PrimitiveType primitive_type>
+void ColumnValueRange<primitive_type>::convert_to_fixed_value() {
     if (!is_fixed_value_convertible()) {
         return;
     }
 
     // Incrementing boolean is denied in C++17, So we use int as bool type
-    using type = std::conditional_t<std::is_same<bool, T>::value, int, T>;
+    using type = std::conditional_t<std::is_same<bool, CppType>::value, int, CppType>;
     type low_value = _low_value;
     type high_value = _high_value;
 
@@ -448,8 +466,8 @@ void ColumnValueRange<T>::convert_to_fixed_value() {
     }
 }
 
-template <class T>
-void ColumnValueRange<T>::convert_to_range_value() {
+template <PrimitiveType primitive_type>
+void ColumnValueRange<primitive_type>::convert_to_range_value() {
     if (!is_range_value_convertible()) {
         return;
     }
@@ -463,8 +481,8 @@ void ColumnValueRange<T>::convert_to_range_value() {
     }
 }
 
-template <class T>
-Status ColumnValueRange<T>::add_range(SQLFilterOp op, T value) {
+template <PrimitiveType primitive_type>
+Status ColumnValueRange<primitive_type>::add_range(SQLFilterOp op, CppType value) {
     if (INVALID_TYPE == _column_type) {
         return Status::InternalError("AddRange failed, Invalid type");
     }
@@ -473,7 +491,7 @@ Status ColumnValueRange<T>::add_range(SQLFilterOp op, T value) {
     _contain_null = false;
 
     if (is_fixed_value_range()) {
-        std::pair<iterator_type, iterator_type> bound_pair = _fixed_values.equal_range(value);
+        std::pair<IteratorType, IteratorType> bound_pair = _fixed_values.equal_range(value);
 
         switch (op) {
         case FILTER_LARGER: {
@@ -564,8 +582,8 @@ Status ColumnValueRange<T>::add_range(SQLFilterOp op, T value) {
     return Status::OK();
 }
 
-template <class T>
-bool ColumnValueRange<T>::is_in_range(const T& value) {
+template <PrimitiveType primitive_type>
+bool ColumnValueRange<primitive_type>::is_in_range(const CppType& value) {
     switch (_high_op) {
     case FILTER_LESS: {
         switch (_low_op) {
@@ -609,8 +627,8 @@ bool ColumnValueRange<T>::is_in_range(const T& value) {
     return false;
 }
 
-template <class T>
-void ColumnValueRange<T>::intersection(ColumnValueRange<T>& range) {
+template <PrimitiveType primitive_type>
+void ColumnValueRange<primitive_type>::intersection(ColumnValueRange<primitive_type>& range) {
     // 1. clear if column type not match
     if (_column_type != range._column_type) {
         set_empty_value_range();
@@ -621,7 +639,7 @@ void ColumnValueRange<T>::intersection(ColumnValueRange<T>& range) {
         set_empty_value_range();
     }
 
-    std::set<T> result_values;
+    std::set<CppType> result_values;
     // 3. fixed_value intersection, fixed value range do not contain null
     if (is_fixed_value_range() || range.is_fixed_value_range()) {
         if (is_fixed_value_range() && range.is_fixed_value_range()) {
@@ -629,7 +647,7 @@ void ColumnValueRange<T>::intersection(ColumnValueRange<T>& range) {
                              range._fixed_values.begin(), range._fixed_values.end(),
                              std::inserter(result_values, result_values.begin()));
         } else if (is_fixed_value_range() && !range.is_fixed_value_range()) {
-            iterator_type iter = _fixed_values.begin();
+            IteratorType iter = _fixed_values.begin();
 
             while (iter != _fixed_values.end()) {
                 if (range.is_in_range(*iter)) {
@@ -638,7 +656,7 @@ void ColumnValueRange<T>::intersection(ColumnValueRange<T>& range) {
                 ++iter;
             }
         } else if (!is_fixed_value_range() && range.is_fixed_value_range()) {
-            iterator_type iter = range._fixed_values.begin();
+            IteratorType iter = range._fixed_values.begin();
             while (iter != range._fixed_values.end()) {
                 if (this->is_in_range(*iter)) {
                     result_values.insert(*iter);
@@ -668,8 +686,8 @@ void ColumnValueRange<T>::intersection(ColumnValueRange<T>& range) {
     }
 }
 
-template <class T>
-bool ColumnValueRange<T>::has_intersection(ColumnValueRange<T>& range) {
+template <PrimitiveType primitive_type>
+bool ColumnValueRange<primitive_type>::has_intersection(ColumnValueRange<primitive_type>& range) {
     // 1. return false if column type not match
     if (_column_type != range._column_type) {
         return false;
@@ -682,7 +700,7 @@ bool ColumnValueRange<T>::has_intersection(ColumnValueRange<T>& range) {
 
     // 3.1 return false if two int fixedRange has no intersection
     if (is_fixed_value_range() && range.is_fixed_value_range()) {
-        std::set<T> result_values;
+        std::set<CppType> result_values;
         set_intersection(_fixed_values.begin(), _fixed_values.end(), range._fixed_values.begin(),
                          range._fixed_values.end(),
                          std::inserter(result_values, result_values.begin()));
@@ -694,7 +712,7 @@ bool ColumnValueRange<T>::has_intersection(ColumnValueRange<T>& range) {
         }
     } // 3.2
     else if (is_fixed_value_range() && !range.is_fixed_value_range()) {
-        iterator_type iter = _fixed_values.begin();
+        IteratorType iter = _fixed_values.begin();
 
         while (iter != _fixed_values.end()) {
             if (range.is_in_range(*iter)) {
@@ -706,7 +724,7 @@ bool ColumnValueRange<T>::has_intersection(ColumnValueRange<T>& range) {
 
         return false;
     } else if (!is_fixed_value_range() && range.is_fixed_value_range()) {
-        iterator_type iter = range._fixed_values.begin();
+        IteratorType iter = range._fixed_values.begin();
 
         while (iter != range._fixed_values.end()) {
             if (this->is_in_range(*iter)) {
@@ -738,11 +756,12 @@ bool ColumnValueRange<T>::has_intersection(ColumnValueRange<T>& range) {
     }
 }
 
-template <class T>
-Status OlapScanKeys::extend_scan_key(ColumnValueRange<T>& range, int32_t max_scan_key_num,
-                                     bool* exact_value) {
+template <PrimitiveType primitive_type>
+Status OlapScanKeys::extend_scan_key(ColumnValueRange<primitive_type>& range,
+                                     int32_t max_scan_key_num, bool* exact_value) {
     using namespace std;
-    using ConstIterator = typename set<T>::const_iterator;
+    using CppType = typename PrimitiveTypeTraits<primitive_type>::CppType;
+    using ConstIterator = typename set<CppType>::const_iterator;
 
     // 1. clear ScanKey if some column range is empty
     if (range.is_empty_value_range()) {
@@ -779,7 +798,7 @@ Status OlapScanKeys::extend_scan_key(ColumnValueRange<T>& range, int32_t max_sca
     if (range.is_fixed_value_range()) {
         // 3.1.1 construct num of fixed value ScanKey (begin_key == end_key)
         if (_begin_scan_keys.empty()) {
-            const set<T>& fixed_value_set = range.get_fixed_value_set();
+            const set<CppType>& fixed_value_set = range.get_fixed_value_set();
             ConstIterator iter = fixed_value_set.begin();
 
             for (; iter != fixed_value_set.end(); ++iter) {
@@ -797,7 +816,7 @@ Status OlapScanKeys::extend_scan_key(ColumnValueRange<T>& range, int32_t max_sca
             }
         } // 3.1.2 produces the Cartesian product of ScanKey and fixed_value
         else {
-            const set<T>& fixed_value_set = range.get_fixed_value_set();
+            const set<CppType>& fixed_value_set = range.get_fixed_value_set();
             int original_key_range_size = _begin_scan_keys.size();
 
             for (int i = 0; i < original_key_range_size; ++i) {
diff --git a/be/src/exec/olap_scan_node.cpp b/be/src/exec/olap_scan_node.cpp
index 846e927a17..f444f36293 100644
--- a/be/src/exec/olap_scan_node.cpp
+++ b/be/src/exec/olap_scan_node.cpp
@@ -574,67 +574,81 @@ Status OlapScanNode::normalize_conjuncts() {
     for (int slot_idx = 0; slot_idx < slots.size(); ++slot_idx) {
         switch (slots[slot_idx]->type().type) {
         case TYPE_TINYINT: {
-            ColumnValueRange<int8_t> range(slots[slot_idx]->col_name(),
-                                           slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_TINYINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_SMALLINT: {
-            ColumnValueRange<int16_t> range(slots[slot_idx]->col_name(),
-                                            slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_SMALLINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_INT: {
-            ColumnValueRange<int32_t> range(slots[slot_idx]->col_name(),
-                                            slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_INT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_BIGINT: {
-            ColumnValueRange<int64_t> range(slots[slot_idx]->col_name(),
-                                            slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_BIGINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_LARGEINT: {
-            ColumnValueRange<__int128> range(slots[slot_idx]->col_name(),
-                                             slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_LARGEINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
-        case TYPE_CHAR:
-        case TYPE_VARCHAR:
-        case TYPE_HLL:
+        case TYPE_CHAR: {
+            ColumnValueRange<TYPE_CHAR> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
+        case TYPE_VARCHAR: {
+            ColumnValueRange<TYPE_VARCHAR> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
+        case TYPE_HLL: {
+            ColumnValueRange<TYPE_HLL> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
         case TYPE_STRING: {
-            ColumnValueRange<StringValue> range(slots[slot_idx]->col_name(),
-                                                slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_STRING> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
-        case TYPE_DATE:
+        case TYPE_DATE: {
+            ColumnValueRange<TYPE_DATE> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
         case TYPE_DATETIME: {
-            ColumnValueRange<DateTimeValue> range(slots[slot_idx]->col_name(),
-                                                  slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_DATETIME> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
+
+        case TYPE_DATEV2: {
+            ColumnValueRange<TYPE_DATEV2> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_DECIMALV2: {
-            ColumnValueRange<DecimalV2Value> range(slots[slot_idx]->col_name(),
-                                                   slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_DECIMALV2> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_BOOLEAN: {
-            ColumnValueRange<bool> range(slots[slot_idx]->col_name(), slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_BOOLEAN> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
@@ -879,7 +893,7 @@ Status OlapScanNode::start_scan_thread(RuntimeState* state) {
     return Status::OK();
 }
 
-template <class T>
+template <PrimitiveType T>
 Status OlapScanNode::normalize_predicate(ColumnValueRange<T>& range, SlotDescriptor* slot) {
     // 1. Normalize InPredicate, add to ColumnValueRange
     RETURN_IF_ERROR(normalize_in_and_eq_predicate(slot, &range));
@@ -994,16 +1008,18 @@ std::pair<bool, void*> OlapScanNode::should_push_down_eq_predicate(doris::SlotDe
     return result_pair;
 }
 
-template <typename T, typename ChangeFixedValueRangeFunc>
-Status OlapScanNode::change_fixed_value_range(ColumnValueRange<T>& temp_range, PrimitiveType type,
+template <PrimitiveType primitive_type, typename ChangeFixedValueRangeFunc>
+Status OlapScanNode::change_fixed_value_range(ColumnValueRange<primitive_type>& temp_range,
                                               void* value, const ChangeFixedValueRangeFunc& func) {
-    switch (type) {
+    switch (primitive_type) {
     case TYPE_DATE: {
         DateTimeValue date_value = *reinterpret_cast<DateTimeValue*>(value);
         // There is must return empty data in olap_scan_node,
         // Because data value loss accuracy
         if (!date_value.check_loss_accuracy_cast_to_date()) {
-            func(temp_range, reinterpret_cast<T*>(&date_value));
+            func(temp_range,
+                 reinterpret_cast<typename PrimitiveTypeTraits<primitive_type>::CppType*>(
+                         &date_value));
         }
         break;
     }
@@ -1018,16 +1034,19 @@ Status OlapScanNode::change_fixed_value_range(ColumnValueRange<T>& temp_range, P
     case TYPE_BIGINT:
     case TYPE_LARGEINT:
     case TYPE_STRING: {
-        func(temp_range, reinterpret_cast<T*>(value));
+        func(temp_range,
+             reinterpret_cast<typename PrimitiveTypeTraits<primitive_type>::CppType*>(value));
         break;
     }
     case TYPE_BOOLEAN: {
         bool v = *reinterpret_cast<bool*>(value);
-        func(temp_range, reinterpret_cast<T*>(&v));
+        func(temp_range,
+             reinterpret_cast<typename PrimitiveTypeTraits<primitive_type>::CppType*>(&v));
         break;
     }
     default: {
-        LOG(WARNING) << "Normalize filter fail, Unsupported Primitive type. [type=" << type << "]";
+        LOG(WARNING) << "Normalize filter fail, Unsupported Primitive type. [type="
+                     << primitive_type << "]";
         return Status::InternalError("Normalize filter fail, Unsupported Primitive type");
     }
     }
@@ -1038,13 +1057,13 @@ Status OlapScanNode::change_fixed_value_range(ColumnValueRange<T>& temp_range, P
 // It will only handle the InPredicate and eq BinaryPredicate in _conjunct_ctxs.
 // It will try to push down conditions of that column as much as possible,
 // But if the number of conditions exceeds the limit, none of conditions will be pushed down.
-template <class T>
+template <PrimitiveType T>
 Status OlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
                                                    ColumnValueRange<T>* range) {
     std::vector<uint32_t> filter_conjuncts_index;
     for (int conj_idx = 0; conj_idx < _conjunct_ctxs.size(); ++conj_idx) {
         // create empty range as temp range, temp range should do intersection on range
-        auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(range->type());
+        auto temp_range = ColumnValueRange<T>::create_empty_column_value_range();
 
         // 1. Normalize in conjuncts like 'where col in (v1, v2, v3)'
         if (TExprOpcode::FILTER_IN == _conjunct_ctxs[conj_idx]->root()->op()) {
@@ -1062,9 +1081,8 @@ Status OlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
                     continue;
                 }
                 auto value = const_cast<void*>(iter->get_value());
-                RETURN_IF_ERROR(
-                        change_fixed_value_range(temp_range, slot->type().type, value,
-                                                 ColumnValueRange<T>::add_fixed_value_range));
+                RETURN_IF_ERROR(change_fixed_value_range(
+                        temp_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 iter->next();
             }
 
@@ -1091,9 +1109,8 @@ Status OlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
                 auto value = result_pair.second;
                 // where A = nullptr should return empty result set
                 if (value != nullptr) {
-                    RETURN_IF_ERROR(
-                            change_fixed_value_range(temp_range, slot->type().type, value,
-                                                     ColumnValueRange<T>::add_fixed_value_range));
+                    RETURN_IF_ERROR(change_fixed_value_range(
+                            temp_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 }
 
                 if (is_key_column(slot->col_name())) {
@@ -1118,14 +1135,13 @@ Status OlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
 // It will only handle the NotInPredicate and not eq BinaryPredicate in _conjunct_ctxs.
 // It will try to push down conditions of that column as much as possible,
 // But if the number of conditions exceeds the limit, none of conditions will be pushed down.
-template <class T>
+template <PrimitiveType T>
 Status OlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot,
                                                            ColumnValueRange<T>* range) {
     // If the conjunct of slot is fixed value, will change the fixed value set of column value range
     // else add value to not in range and push down predicate directly
     bool is_fixed_range = range->is_fixed_value_range();
-    auto not_in_range = ColumnValueRange<T>::create_empty_column_value_range(range->column_name(),
-                                                                             range->type());
+    auto not_in_range = ColumnValueRange<T>::create_empty_column_value_range(range->column_name());
 
     std::vector<uint32_t> filter_conjuncts_index;
     for (int conj_idx = 0; conj_idx < _conjunct_ctxs.size(); ++conj_idx) {
@@ -1146,12 +1162,10 @@ Status OlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot,
                 auto value = const_cast<void*>(iter->get_value());
                 if (is_fixed_range) {
                     RETURN_IF_ERROR(change_fixed_value_range(
-                            *range, slot->type().type, value,
-                            ColumnValueRange<T>::remove_fixed_value_range));
+                            *range, value, ColumnValueRange<T>::remove_fixed_value_range));
                 } else {
-                    RETURN_IF_ERROR(
-                            change_fixed_value_range(not_in_range, slot->type().type, value,
-                                                     ColumnValueRange<T>::add_fixed_value_range));
+                    RETURN_IF_ERROR(change_fixed_value_range(
+                            not_in_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 }
                 iter->next();
             }
@@ -1181,12 +1195,10 @@ Status OlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot,
 
                 if (is_fixed_range) {
                     RETURN_IF_ERROR(change_fixed_value_range(
-                            *range, slot->type().type, value,
-                            ColumnValueRange<T>::remove_fixed_value_range));
+                            *range, value, ColumnValueRange<T>::remove_fixed_value_range));
                 } else {
-                    RETURN_IF_ERROR(
-                            change_fixed_value_range(not_in_range, slot->type().type, value,
-                                                     ColumnValueRange<T>::add_fixed_value_range));
+                    RETURN_IF_ERROR(change_fixed_value_range(
+                            not_in_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 }
 
                 if (is_key_column(slot->col_name())) {
@@ -1209,7 +1221,7 @@ Status OlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot,
     return Status::OK();
 }
 
-template <typename T>
+template <PrimitiveType T>
 bool OlapScanNode::normalize_is_null_predicate(Expr* expr, SlotDescriptor* slot,
                                                const std::string& is_null_str,
                                                ColumnValueRange<T>* range) {
@@ -1226,14 +1238,14 @@ bool OlapScanNode::normalize_is_null_predicate(Expr* expr, SlotDescriptor* slot,
         return false;
     }
 
-    auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(range->type());
+    auto temp_range = ColumnValueRange<T>::create_empty_column_value_range();
     temp_range.set_contain_null(is_null_str == "null");
     range->intersection(temp_range);
 
     return true;
 }
 
-template <class T>
+template <PrimitiveType T>
 Status OlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
                                                       ColumnValueRange<T>* range) {
     std::vector<uint32_t> filter_conjuncts_index;
@@ -1303,7 +1315,8 @@ Status OlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
                         }
                     }
                     range->add_range(to_olap_filter_type(pred->op(), child_idx),
-                                     *reinterpret_cast<T*>(&date_value));
+                                     *reinterpret_cast<typename PrimitiveTypeTraits<T>::CppType*>(
+                                             &date_value));
                     break;
                 }
                 case TYPE_TINYINT:
@@ -1318,8 +1331,9 @@ Status OlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
                 case TYPE_LARGEINT:
                 case TYPE_BOOLEAN:
                 case TYPE_STRING: {
-                    range->add_range(to_olap_filter_type(pred->op(), child_idx),
-                                     *reinterpret_cast<T*>(value));
+                    range->add_range(
+                            to_olap_filter_type(pred->op(), child_idx),
+                            *reinterpret_cast<typename PrimitiveTypeTraits<T>::CppType*>(value));
                     break;
                 }
 
@@ -1337,7 +1351,9 @@ Status OlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
 
                 VLOG_CRITICAL << slot->col_name() << " op: "
                               << static_cast<int>(to_olap_filter_type(pred->op(), child_idx))
-                              << " value: " << *reinterpret_cast<T*>(value);
+                              << " value: "
+                              << *reinterpret_cast<typename PrimitiveTypeTraits<T>::CppType*>(
+                                         value);
             }
         }
     }
diff --git a/be/src/exec/olap_scan_node.h b/be/src/exec/olap_scan_node.h
index 88c17ba1e2..47eee0e53d 100644
--- a/be/src/exec/olap_scan_node.h
+++ b/be/src/exec/olap_scan_node.h
@@ -110,21 +110,21 @@ protected:
     Status build_key_ranges_and_filters();
     Status start_scan_thread(RuntimeState* state);
 
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_predicate(ColumnValueRange<T>& range, SlotDescriptor* slot);
 
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_in_and_eq_predicate(SlotDescriptor* slot, ColumnValueRange<T>* range);
 
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot, ColumnValueRange<T>* range);
 
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_noneq_binary_predicate(SlotDescriptor* slot, ColumnValueRange<T>* range);
 
     Status normalize_bloom_filter_predicate(SlotDescriptor* slot);
 
-    template <typename T>
+    template <PrimitiveType T>
     static bool normalize_is_null_predicate(Expr* expr, SlotDescriptor* slot,
                                             const std::string& is_null_str,
                                             ColumnValueRange<T>* range);
@@ -148,9 +148,9 @@ protected:
 
     bool should_push_down_in_predicate(SlotDescriptor* slot, InPredicate* in_pred);
 
-    template <typename T, typename ChangeFixedValueRangeFunc>
-    static Status change_fixed_value_range(ColumnValueRange<T>& range, PrimitiveType type,
-                                           void* value, const ChangeFixedValueRangeFunc& func);
+    template <PrimitiveType T, typename ChangeFixedValueRangeFunc>
+    static Status change_fixed_value_range(ColumnValueRange<T>& range, void* value,
+                                           const ChangeFixedValueRangeFunc& func);
 
     std::pair<bool, void*> should_push_down_eq_predicate(SlotDescriptor* slot, Expr* pred,
                                                          int conj_idx, int child_idx);
diff --git a/be/src/runtime/primitive_type.h b/be/src/runtime/primitive_type.h
index b2ed96592d..35b4b3160a 100644
--- a/be/src/runtime/primitive_type.h
+++ b/be/src/runtime/primitive_type.h
@@ -175,6 +175,12 @@ struct PrimitiveTypeTraits<TYPE_STRING> {
     using ColumnType = vectorized::ColumnString;
 };
 
+template <>
+struct PrimitiveTypeTraits<TYPE_HLL> {
+    using CppType = StringValue;
+    using ColumnType = vectorized::ColumnString;
+};
+
 // only for adapt get_predicate_column_ptr
 template <PrimitiveType type>
 struct PredicatePrimitiveTypeTraits {
diff --git a/be/src/vec/exec/volap_scan_node.cpp b/be/src/vec/exec/volap_scan_node.cpp
index fd77f50ff7..22382f5ce6 100644
--- a/be/src/vec/exec/volap_scan_node.cpp
+++ b/be/src/vec/exec/volap_scan_node.cpp
@@ -642,74 +642,81 @@ Status VOlapScanNode::normalize_conjuncts() {
     for (int slot_idx = 0; slot_idx < slots.size(); ++slot_idx) {
         switch (slots[slot_idx]->type().type) {
         case TYPE_TINYINT: {
-            ColumnValueRange<int8_t> range(slots[slot_idx]->col_name(),
-                                           slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_TINYINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_SMALLINT: {
-            ColumnValueRange<int16_t> range(slots[slot_idx]->col_name(),
-                                            slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_SMALLINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_INT: {
-            ColumnValueRange<int32_t> range(slots[slot_idx]->col_name(),
-                                            slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_INT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_BIGINT: {
-            ColumnValueRange<int64_t> range(slots[slot_idx]->col_name(),
-                                            slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_BIGINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_LARGEINT: {
-            ColumnValueRange<__int128> range(slots[slot_idx]->col_name(),
-                                             slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_LARGEINT> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
-        case TYPE_CHAR:
-        case TYPE_VARCHAR:
-        case TYPE_HLL:
+        case TYPE_CHAR: {
+            ColumnValueRange<TYPE_CHAR> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
+        case TYPE_VARCHAR: {
+            ColumnValueRange<TYPE_VARCHAR> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
+        case TYPE_HLL: {
+            ColumnValueRange<TYPE_HLL> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
         case TYPE_STRING: {
-            ColumnValueRange<StringValue> range(slots[slot_idx]->col_name(),
-                                                slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_STRING> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
-        case TYPE_DATE:
+        case TYPE_DATE: {
+            ColumnValueRange<TYPE_DATE> range(slots[slot_idx]->col_name());
+            normalize_predicate(range, slots[slot_idx]);
+            break;
+        }
         case TYPE_DATETIME: {
-            ColumnValueRange<DateTimeValue> range(slots[slot_idx]->col_name(),
-                                                  slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_DATETIME> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_DATEV2: {
-            ColumnValueRange<doris::vectorized::DateV2Value> range(slots[slot_idx]->col_name(),
-                                                                   slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_DATEV2> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_DECIMALV2: {
-            ColumnValueRange<DecimalV2Value> range(slots[slot_idx]->col_name(),
-                                                   slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_DECIMALV2> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
 
         case TYPE_BOOLEAN: {
-            ColumnValueRange<bool> range(slots[slot_idx]->col_name(), slots[slot_idx]->type().type);
+            ColumnValueRange<TYPE_BOOLEAN> range(slots[slot_idx]->col_name());
             normalize_predicate(range, slots[slot_idx]);
             break;
         }
@@ -830,7 +837,7 @@ Status VOlapScanNode::start_scan(RuntimeState* state) {
     return Status::OK();
 }
 
-template <class T>
+template <PrimitiveType T>
 Status VOlapScanNode::normalize_predicate(ColumnValueRange<T>& range, SlotDescriptor* slot) {
     // 1. Normalize InPredicate, add to ColumnValueRange
     RETURN_IF_ERROR(normalize_in_and_eq_predicate(slot, &range));
@@ -946,16 +953,18 @@ std::pair<bool, void*> VOlapScanNode::should_push_down_eq_predicate(doris::SlotD
     return result_pair;
 }
 
-template <typename T, typename ChangeFixedValueRangeFunc>
-Status VOlapScanNode::change_fixed_value_range(ColumnValueRange<T>& temp_range, PrimitiveType type,
+template <PrimitiveType primitive_type, typename ChangeFixedValueRangeFunc>
+Status VOlapScanNode::change_fixed_value_range(ColumnValueRange<primitive_type>& temp_range,
                                                void* value, const ChangeFixedValueRangeFunc& func) {
-    switch (type) {
+    switch (primitive_type) {
     case TYPE_DATE: {
         DateTimeValue date_value = *reinterpret_cast<DateTimeValue*>(value);
         // There is must return empty data in olap_scan_node,
         // Because data value loss accuracy
         if (!date_value.check_loss_accuracy_cast_to_date()) {
-            func(temp_range, reinterpret_cast<T*>(&date_value));
+            func(temp_range,
+                 reinterpret_cast<typename PrimitiveTypeTraits<primitive_type>::CppType*>(
+                         &date_value));
         }
         break;
     }
@@ -970,12 +979,14 @@ Status VOlapScanNode::change_fixed_value_range(ColumnValueRange<T>& temp_range,
     case TYPE_BIGINT:
     case TYPE_LARGEINT:
     case TYPE_STRING: {
-        func(temp_range, reinterpret_cast<T*>(value));
+        func(temp_range,
+             reinterpret_cast<typename PrimitiveTypeTraits<primitive_type>::CppType*>(value));
         break;
     }
     case TYPE_BOOLEAN: {
         bool v = *reinterpret_cast<bool*>(value);
-        func(temp_range, reinterpret_cast<T*>(&v));
+        func(temp_range,
+             reinterpret_cast<typename PrimitiveTypeTraits<primitive_type>::CppType*>(&v));
         break;
     }
     case TYPE_DATEV2: {
@@ -983,7 +994,7 @@ Status VOlapScanNode::change_fixed_value_range(ColumnValueRange<T>& temp_range,
         if (!date_value.check_loss_accuracy_cast_to_date()) {
             doris::vectorized::DateV2Value date_v2;
             date_v2.convert_dt_to_date_v2(&date_value);
-            if constexpr (std::is_same_v<T, doris::vectorized::DateV2Value>) {
+            if constexpr (primitive_type == PrimitiveType::TYPE_DATEV2) {
                 func(temp_range, &date_v2);
             } else {
                 __builtin_unreachable();
@@ -992,7 +1003,8 @@ Status VOlapScanNode::change_fixed_value_range(ColumnValueRange<T>& temp_range,
         break;
     }
     default: {
-        LOG(WARNING) << "Normalize filter fail, Unsupported Primitive type. [type=" << type << "]";
+        LOG(WARNING) << "Normalize filter fail, Unsupported Primitive type. [type="
+                     << primitive_type << "]";
         return Status::InternalError("Normalize filter fail, Unsupported Primitive type");
     }
     }
@@ -1066,13 +1078,13 @@ void VOlapScanNode::remove_pushed_conjuncts(RuntimeState* state) {
 // It will only handle the InPredicate and eq BinaryPredicate in conjunct_ctxs.
 // It will try to push down conditions of that column as much as possible,
 // But if the number of conditions exceeds the limit, none of conditions will be pushed down.
-template <class T>
+template <PrimitiveType T>
 Status VOlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
                                                     ColumnValueRange<T>* range) {
     std::vector<uint32_t> filter_conjuncts_index;
     for (int conj_idx = 0; conj_idx < _conjunct_ctxs.size(); ++conj_idx) {
         // create empty range as temp range, temp range should do intersection on range
-        auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(range->type());
+        auto temp_range = ColumnValueRange<T>::create_empty_column_value_range();
 
         // 1. Normalize in conjuncts like 'where col in (v1, v2, v3)'
         if (TExprOpcode::FILTER_IN == _conjunct_ctxs[conj_idx]->root()->op()) {
@@ -1090,9 +1102,8 @@ Status VOlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
                     continue;
                 }
                 auto value = const_cast<void*>(iter->get_value());
-                RETURN_IF_ERROR(
-                        change_fixed_value_range(temp_range, slot->type().type, value,
-                                                 ColumnValueRange<T>::add_fixed_value_range));
+                RETURN_IF_ERROR(change_fixed_value_range(
+                        temp_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 iter->next();
             }
 
@@ -1119,9 +1130,8 @@ Status VOlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
                 auto value = result_pair.second;
                 // where A = nullptr should return empty result set
                 if (value != nullptr) {
-                    RETURN_IF_ERROR(
-                            change_fixed_value_range(temp_range, slot->type().type, value,
-                                                     ColumnValueRange<T>::add_fixed_value_range));
+                    RETURN_IF_ERROR(change_fixed_value_range(
+                            temp_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 }
 
                 if (is_key_column(slot->col_name())) {
@@ -1146,14 +1156,13 @@ Status VOlapScanNode::normalize_in_and_eq_predicate(SlotDescriptor* slot,
 // It will only handle the NotInPredicate and not eq BinaryPredicate in conjunct_ctxs.
 // It will try to push down conditions of that column as much as possible,
 // But if the number of conditions exceeds the limit, none of conditions will be pushed down.
-template <class T>
+template <PrimitiveType T>
 Status VOlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot,
                                                             ColumnValueRange<T>* range) {
     // If the conjunct of slot is fixed value, will change the fixed value set of column value range
     // else add value to not in range and push down predicate directly
     bool is_fixed_range = range->is_fixed_value_range();
-    auto not_in_range = ColumnValueRange<T>::create_empty_column_value_range(range->column_name(),
-                                                                             range->type());
+    auto not_in_range = ColumnValueRange<T>::create_empty_column_value_range(range->column_name());
 
     std::vector<uint32_t> filter_conjuncts_index;
     for (int conj_idx = 0; conj_idx < _conjunct_ctxs.size(); ++conj_idx) {
@@ -1174,12 +1183,10 @@ Status VOlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot
                 auto value = const_cast<void*>(iter->get_value());
                 if (is_fixed_range) {
                     RETURN_IF_ERROR(change_fixed_value_range(
-                            *range, slot->type().type, value,
-                            ColumnValueRange<T>::remove_fixed_value_range));
+                            *range, value, ColumnValueRange<T>::remove_fixed_value_range));
                 } else {
-                    RETURN_IF_ERROR(
-                            change_fixed_value_range(not_in_range, slot->type().type, value,
-                                                     ColumnValueRange<T>::add_fixed_value_range));
+                    RETURN_IF_ERROR(change_fixed_value_range(
+                            not_in_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 }
                 iter->next();
             }
@@ -1209,12 +1216,10 @@ Status VOlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot
 
                 if (is_fixed_range) {
                     RETURN_IF_ERROR(change_fixed_value_range(
-                            *range, slot->type().type, value,
-                            ColumnValueRange<T>::remove_fixed_value_range));
+                            *range, value, ColumnValueRange<T>::remove_fixed_value_range));
                 } else {
-                    RETURN_IF_ERROR(
-                            change_fixed_value_range(not_in_range, slot->type().type, value,
-                                                     ColumnValueRange<T>::add_fixed_value_range));
+                    RETURN_IF_ERROR(change_fixed_value_range(
+                            not_in_range, value, ColumnValueRange<T>::add_fixed_value_range));
                 }
 
                 if (is_key_column(slot->col_name())) {
@@ -1237,7 +1242,7 @@ Status VOlapScanNode::normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot
     return Status::OK();
 }
 
-template <typename T>
+template <PrimitiveType T>
 bool VOlapScanNode::normalize_is_null_predicate(Expr* expr, SlotDescriptor* slot,
                                                 const std::string& is_null_str,
                                                 ColumnValueRange<T>* range) {
@@ -1254,14 +1259,14 @@ bool VOlapScanNode::normalize_is_null_predicate(Expr* expr, SlotDescriptor* slot
         return false;
     }
 
-    auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(range->type());
+    auto temp_range = ColumnValueRange<T>::create_empty_column_value_range();
     temp_range.set_contain_null(is_null_str == "null");
     range->intersection(temp_range);
 
     return true;
 }
 
-template <class T>
+template <PrimitiveType T>
 Status VOlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
                                                        ColumnValueRange<T>* range) {
     std::vector<uint32_t> filter_conjuncts_index;
@@ -1331,7 +1336,8 @@ Status VOlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
                         }
                     }
                     range->add_range(to_olap_filter_type(pred->op(), child_idx),
-                                     *reinterpret_cast<T*>(&date_value));
+                                     *reinterpret_cast<typename PrimitiveTypeTraits<T>::CppType*>(
+                                             &date_value));
                     break;
                 }
                 case TYPE_DATEV2: {
@@ -1343,7 +1349,7 @@ Status VOlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
                     }
                     doris::vectorized::DateV2Value date_v2;
                     date_v2.convert_dt_to_date_v2(&date_value);
-                    if constexpr (std::is_same_v<T, doris::vectorized::DateV2Value>) {
+                    if constexpr (T == PrimitiveType::TYPE_DATEV2) {
                         range->add_range(to_olap_filter_type(pred->op(), child_idx), date_v2);
                         break;
                     } else {
@@ -1362,8 +1368,9 @@ Status VOlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
                 case TYPE_LARGEINT:
                 case TYPE_BOOLEAN:
                 case TYPE_STRING: {
-                    range->add_range(to_olap_filter_type(pred->op(), child_idx),
-                                     *reinterpret_cast<T*>(value));
+                    range->add_range(
+                            to_olap_filter_type(pred->op(), child_idx),
+                            *reinterpret_cast<typename PrimitiveTypeTraits<T>::CppType*>(value));
                     break;
                 }
 
@@ -1381,7 +1388,9 @@ Status VOlapScanNode::normalize_noneq_binary_predicate(SlotDescriptor* slot,
 
                 VLOG_CRITICAL << slot->col_name() << " op: "
                               << static_cast<int>(to_olap_filter_type(pred->op(), child_idx))
-                              << " value: " << *reinterpret_cast<T*>(value);
+                              << " value: "
+                              << *reinterpret_cast<typename PrimitiveTypeTraits<T>::CppType*>(
+                                         value);
             }
         }
     }
diff --git a/be/src/vec/exec/volap_scan_node.h b/be/src/vec/exec/volap_scan_node.h
index 343f965c69..29ea94f94e 100644
--- a/be/src/vec/exec/volap_scan_node.h
+++ b/be/src/vec/exec/volap_scan_node.h
@@ -69,29 +69,29 @@ private:
     void eval_const_conjuncts();
     Status normalize_conjuncts();
     Status build_key_ranges_and_filters();
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_predicate(ColumnValueRange<T>& range, SlotDescriptor* slot);
 
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_in_and_eq_predicate(SlotDescriptor* slot, ColumnValueRange<T>* range);
 
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_not_in_and_not_eq_predicate(SlotDescriptor* slot, ColumnValueRange<T>* range);
 
-    template <class T>
+    template <PrimitiveType T>
     Status normalize_noneq_binary_predicate(SlotDescriptor* slot, ColumnValueRange<T>* range);
 
     Status normalize_bloom_filter_predicate(SlotDescriptor* slot);
 
-    template <typename T>
+    template <PrimitiveType T>
     static bool normalize_is_null_predicate(Expr* expr, SlotDescriptor* slot,
                                             const std::string& is_null_str,
                                             ColumnValueRange<T>* range);
     bool should_push_down_in_predicate(SlotDescriptor* slot, InPredicate* in_pred);
 
-    template <typename T, typename ChangeFixedValueRangeFunc>
-    static Status change_fixed_value_range(ColumnValueRange<T>& range, PrimitiveType type,
-                                           void* value, const ChangeFixedValueRangeFunc& func);
+    template <PrimitiveType T, typename ChangeFixedValueRangeFunc>
+    static Status change_fixed_value_range(ColumnValueRange<T>& range, void* value,
+                                           const ChangeFixedValueRangeFunc& func);
 
     std::pair<bool, void*> should_push_down_eq_predicate(SlotDescriptor* slot, Expr* pred,
                                                          int conj_idx, int child_idx);
diff --git a/be/test/exec/olap_common_test.cpp b/be/test/exec/olap_common_test.cpp
index d444f2f7b3..68324e06f6 100644
--- a/be/test/exec/olap_common_test.cpp
+++ b/be/test/exec/olap_common_test.cpp
@@ -62,13 +62,13 @@ public:
 };
 
 TEST_F(ColumnValueRangeTest, ExceptionCase) {
-    ColumnValueRange<int32_t> range1;
+    ColumnValueRange<TYPE_INT> range1;
     EXPECT_FALSE(range1.add_fixed_value(10).ok());
     EXPECT_FALSE(range1.add_range(FILTER_LESS_OR_EQUAL, 10).ok());
 }
 
 TEST_F(ColumnValueRangeTest, NormalCase) {
-    ColumnValueRange<int32_t> range1("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range1("col");
 
     EXPECT_TRUE(range1.add_fixed_value(10).ok());
     EXPECT_TRUE(range1.add_fixed_value(20).ok());
@@ -79,7 +79,7 @@ TEST_F(ColumnValueRangeTest, NormalCase) {
     EXPECT_TRUE(range1.add_range(FILTER_LESS, 30).ok());
     EXPECT_FALSE(range1.is_empty_value_range());
 
-    ColumnValueRange<int32_t> range2("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range2("col");
     EXPECT_TRUE(range2.add_fixed_value(30).ok());
     EXPECT_FALSE(range1.has_intersection(range2));
 
@@ -95,7 +95,7 @@ TEST_F(ColumnValueRangeTest, NormalCase) {
 }
 
 TEST_F(ColumnValueRangeTest, FixedAddRangeTest) {
-    ColumnValueRange<int32_t> range1("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range1("col");
 
     for (int i = 0; i < 100; i += 10) {
         EXPECT_TRUE(range1.add_fixed_value(i).ok());
@@ -152,20 +152,20 @@ TEST_F(ColumnValueRangeTest, FixedAddRangeTest) {
 }
 
 TEST_F(ColumnValueRangeTest, ContainsNullTest) {
-    ColumnValueRange<int32_t> range1("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range1("col");
 
     // test fixed value range intersection with null and no null range
     for (int i = 0; i < 100; i += 10) {
         EXPECT_TRUE(range1.add_fixed_value(i).ok());
     }
 
-    auto null_range = ColumnValueRange<int32_t>::create_empty_column_value_range(TYPE_INT);
+    auto null_range = ColumnValueRange<TYPE_INT>::create_empty_column_value_range();
     null_range.set_contain_null(true);
     EXPECT_TRUE(!null_range.is_empty_value_range());
     null_range.intersection(range1);
     EXPECT_TRUE(null_range.is_empty_value_range());
 
-    auto no_null_range = ColumnValueRange<int32_t>::create_empty_column_value_range(TYPE_INT);
+    auto no_null_range = ColumnValueRange<TYPE_INT>::create_empty_column_value_range();
     no_null_range.set_contain_null(false);
     no_null_range.intersection(range1);
     EXPECT_EQ(no_null_range._fixed_values, range1._fixed_values);
@@ -176,13 +176,13 @@ TEST_F(ColumnValueRangeTest, ContainsNullTest) {
     range1.add_range(FILTER_LESS_OR_EQUAL, 80);
     range1.add_range(FILTER_LARGER, 50);
 
-    null_range = ColumnValueRange<int32_t>::create_empty_column_value_range(TYPE_INT);
+    null_range = ColumnValueRange<TYPE_INT>::create_empty_column_value_range();
     null_range.set_contain_null(true);
     EXPECT_TRUE(!null_range.is_empty_value_range());
     null_range.intersection(range1);
     EXPECT_TRUE(null_range.is_empty_value_range());
 
-    no_null_range = ColumnValueRange<int32_t>::create_empty_column_value_range(TYPE_INT);
+    no_null_range = ColumnValueRange<TYPE_INT>::create_empty_column_value_range();
     no_null_range.set_contain_null(false);
     no_null_range.intersection(range1);
     EXPECT_TRUE(no_null_range._fixed_values.empty());
@@ -192,7 +192,7 @@ TEST_F(ColumnValueRangeTest, ContainsNullTest) {
 }
 
 TEST_F(ColumnValueRangeTest, RangeAddRangeTest) {
-    ColumnValueRange<int32_t> range1("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range1("col");
 
     EXPECT_EQ(range1.get_range_min_value(), std::numeric_limits<int32_t>::min());
     EXPECT_EQ(range1.get_range_max_value(), std::numeric_limits<int32_t>::max());
@@ -220,10 +220,10 @@ TEST_F(ColumnValueRangeTest, RangeAddRangeTest) {
 }
 
 TEST_F(ColumnValueRangeTest, RangeIntersectionTest) {
-    ColumnValueRange<int32_t> range1("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range1("col");
     EXPECT_TRUE(range1.add_range(FILTER_LARGER_OR_EQUAL, 20).ok());
 
-    ColumnValueRange<int32_t> range2("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range2("col");
     EXPECT_TRUE(range2.add_range(FILTER_LESS, 100).ok());
 
     EXPECT_TRUE(range1.has_intersection(range2));
@@ -256,13 +256,13 @@ TEST_F(ColumnValueRangeTest, RangeIntersectionTest) {
 }
 
 TEST_F(ColumnValueRangeTest, FixedValueIntersectionTest) {
-    ColumnValueRange<int32_t> range1("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range1("col");
 
     for (int i = 0; i < 100; i += 10) {
         EXPECT_TRUE(range1.add_fixed_value(i).ok());
     }
 
-    ColumnValueRange<int32_t> range2("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range2("col");
 
     for (int i = 50; i < 200; i += 10) {
         EXPECT_TRUE(range2.add_fixed_value(i).ok());
@@ -310,8 +310,8 @@ TEST_F(ColumnValueRangeTest, FixedAndRangeIntersectionTest) {
     for (int type = TYPE_TINYINT; type <= TYPE_BIGINT; type++) {
         switch (type) {
         case TYPE_TINYINT: {
-            ColumnValueRange<int8_t> range1("col", TYPE_TINYINT);
-            ColumnValueRange<int8_t> range2("col", TYPE_TINYINT);
+            ColumnValueRange<TYPE_TINYINT> range1("col");
+            ColumnValueRange<TYPE_TINYINT> range2("col");
 
             for (int i = 0; i < 100; i += 10) {
                 EXPECT_TRUE(range1.add_fixed_value(i).ok());
@@ -338,8 +338,8 @@ TEST_F(ColumnValueRangeTest, FixedAndRangeIntersectionTest) {
         }
 
         case TYPE_SMALLINT: {
-            ColumnValueRange<int16_t> range1("col", TYPE_SMALLINT);
-            ColumnValueRange<int16_t> range2("col", TYPE_SMALLINT);
+            ColumnValueRange<TYPE_SMALLINT> range1("col");
+            ColumnValueRange<TYPE_SMALLINT> range2("col");
 
             for (int i = 0; i < 100; i += 10) {
                 EXPECT_TRUE(range1.add_fixed_value(i).ok());
@@ -366,8 +366,8 @@ TEST_F(ColumnValueRangeTest, FixedAndRangeIntersectionTest) {
         }
 
         case TYPE_INT: {
-            ColumnValueRange<int32_t> range1("col", TYPE_INT);
-            ColumnValueRange<int32_t> range2("col", TYPE_INT);
+            ColumnValueRange<TYPE_INT> range1("col");
+            ColumnValueRange<TYPE_INT> range2("col");
 
             for (int i = 0; i < 100; i += 10) {
                 EXPECT_TRUE(range1.add_fixed_value(i).ok());
@@ -394,8 +394,8 @@ TEST_F(ColumnValueRangeTest, FixedAndRangeIntersectionTest) {
         }
 
         case TYPE_BIGINT: {
-            ColumnValueRange<int64_t> range1("col", TYPE_BIGINT);
-            ColumnValueRange<int64_t> range2("col", TYPE_BIGINT);
+            ColumnValueRange<TYPE_BIGINT> range1("col");
+            ColumnValueRange<TYPE_BIGINT> range2("col");
 
             for (int i = 0; i < 100; i += 10) {
                 EXPECT_TRUE(range1.add_fixed_value(i).ok());
@@ -437,7 +437,7 @@ public:
 TEST_F(OlapScanKeysTest, ExtendFixedTest) {
     OlapScanKeys scan_keys;
 
-    ColumnValueRange<int32_t> range1("col", TYPE_BIGINT);
+    ColumnValueRange<TYPE_INT> range1("col");
 
     for (int i = 0; i < 3; ++i) {
         EXPECT_TRUE(range1.add_fixed_value(i).ok());
@@ -461,7 +461,7 @@ TEST_F(OlapScanKeysTest, ExtendFixedTest) {
     EXPECT_EQ(OlapScanKeys::to_print_key(key_range[2]->begin_scan_range), "2");
     EXPECT_EQ(OlapScanKeys::to_print_key(key_range[2]->end_scan_range), "2");
 
-    ColumnValueRange<int32_t> range2("col", TYPE_BIGINT);
+    ColumnValueRange<TYPE_INT> range2("col");
 
     for (int i = 0; i < 2; ++i) {
         EXPECT_TRUE(range2.add_fixed_value(i).ok());
@@ -526,7 +526,7 @@ TEST_F(OlapScanKeysTest, ExtendFixedTest) {
 TEST_F(OlapScanKeysTest, ExtendFixedAndRangeTest) {
     OlapScanKeys scan_keys;
 
-    ColumnValueRange<int32_t> range1("col", TYPE_BIGINT);
+    ColumnValueRange<TYPE_INT> range1("col");
 
     for (int i = 0; i < 3; ++i) {
         EXPECT_TRUE(range1.add_fixed_value(i).ok());
@@ -536,7 +536,7 @@ TEST_F(OlapScanKeysTest, ExtendFixedAndRangeTest) {
     scan_keys.extend_scan_key(range1, 1024, &exact_range);
     EXPECT_EQ(exact_range, true);
 
-    ColumnValueRange<int32_t> range2("col", TYPE_BIGINT);
+    ColumnValueRange<TYPE_INT> range2("col");
     EXPECT_TRUE(range2.add_range(FILTER_LARGER_OR_EQUAL, 20).ok());
 
     exact_range = true;
@@ -582,7 +582,7 @@ TEST_F(OlapScanKeysTest, ExtendRangeTest) {
     OlapScanKeys scan_keys;
     config::doris_max_scan_key_num = 1;
 
-    ColumnValueRange<int64_t> range2("col", TYPE_BIGINT);
+    ColumnValueRange<TYPE_BIGINT> range2("col");
     EXPECT_TRUE(range2.add_range(FILTER_LARGER_OR_EQUAL, 20).ok());
     EXPECT_TRUE(range2.add_range(FILTER_LESS_OR_EQUAL, 100).ok());
 
@@ -619,7 +619,7 @@ TEST_F(OlapScanKeysTest, EachtypeTest) {
 
     {
         OlapScanKeys scan_keys;
-        ColumnValueRange<int8_t> range("col", TYPE_TINYINT);
+        ColumnValueRange<TYPE_TINYINT> range("col");
         bool exact_range = true;
         EXPECT_TRUE(scan_keys.extend_scan_key(range, 1024, &exact_range).ok());
         EXPECT_EQ(exact_range, true);
@@ -643,7 +643,7 @@ TEST_F(OlapScanKeysTest, EachtypeTest) {
 
     {
         OlapScanKeys scan_keys;
-        ColumnValueRange<int16_t> range("col", TYPE_SMALLINT);
+        ColumnValueRange<TYPE_SMALLINT> range("col");
         bool exact_range = true;
         EXPECT_TRUE(scan_keys.extend_scan_key(range, 1024, &exact_range).ok());
         EXPECT_EQ(exact_range, true);
@@ -677,7 +677,7 @@ TEST_F(OlapScanKeysTest, EachtypeTest) {
 }
 
 TEST_F(OlapScanKeysTest, ToOlapFilterTest) {
-    ColumnValueRange<int32_t> range("col", TYPE_INT);
+    ColumnValueRange<TYPE_INT> range("col");
 
     std::vector<TCondition> filters;
     range.to_olap_filter(filters);