[refactor](type) Use PrimitiveType to replace TypeIndex (PART II) (apache#50700)

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [ ] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: Gabriel39

be/src/olap/rowset/segment_creator.cpp

@@ -93,7 +93,7 @@ Status SegmentFlusher::_internal_parse_variant_columns(vectorized::Block& block)
     std::vector<int> variant_column_pos;
     for (int i = 0; i < block.columns(); ++i) {
         const auto& entry = block.get_by_position(i);
-        if (vectorized::is_variant_type(remove_nullable(entry.type))) {
+        if (entry.type->get_primitive_type() == TYPE_VARIANT) {
             variant_column_pos.push_back(i);
         }
     }

be/src/olap/rowset/segment_v2/segment.h

@@ -204,8 +204,7 @@ class Segment : public std::enable_shared_from_this<Segment>, public MetadataAdd
             // Default column iterator
             return true;
         }
-        if (vectorized::WhichDataType(vectorized::remove_nullable(storage_column_type))
-                    .is_variant_type()) {
+        if (storage_column_type->get_primitive_type() == TYPE_VARIANT) {
             // Predicate should nerver apply on variant type
             return false;
         }

be/src/pipeline/exec/aggregation_source_operator.cpp

@@ -405,7 +405,7 @@ Status AggLocalState::_get_without_key_result(RuntimeState* state, vectorized::B
     for (int i = 0; i < block_schema.size(); ++i) {
         const auto column_type = block_schema[i].type;
         if (!column_type->equals(*data_types[i])) {
-            if (!vectorized::is_array(remove_nullable(column_type))) {
+            if (column_type->get_primitive_type() != TYPE_ARRAY) {
                 if (!column_type->is_nullable() || data_types[i]->is_nullable() ||
                     !remove_nullable(column_type)->equals(*data_types[i])) {
                     return Status::InternalError(

be/src/runtime/primitive_type.h

@@ -100,6 +100,18 @@ constexpr bool is_date_type(PrimitiveType type) {
            type == TYPE_DATEV2;
 }
 
+constexpr bool is_date_or_datetime(PrimitiveType type) {
+    return type == TYPE_DATETIME || type == TYPE_DATE;
+}
+
+constexpr bool is_date_v2_or_datetime_v2(PrimitiveType type) {
+    return type == TYPE_DATETIMEV2 || type == TYPE_DATEV2;
+}
+
+constexpr bool is_ip(PrimitiveType type) {
+    return type == TYPE_IPV4 || type == TYPE_IPV6;
+}
+
 constexpr bool is_string_type(PrimitiveType type) {
     return type == TYPE_CHAR || type == TYPE_VARCHAR || type == TYPE_STRING;
 }
@@ -120,16 +132,24 @@ constexpr bool is_float_or_double(PrimitiveType type) {
     return type == TYPE_FLOAT || type == TYPE_DOUBLE;
 }
 
+constexpr bool is_int(PrimitiveType type) {
+    return type == TYPE_TINYINT || type == TYPE_SMALLINT || type == TYPE_INT ||
+           type == TYPE_BIGINT || type == TYPE_LARGEINT;
+}
+
 constexpr bool is_int_or_bool(PrimitiveType type) {
-    return type == TYPE_BOOLEAN || type == TYPE_TINYINT || type == TYPE_SMALLINT ||
-           type == TYPE_INT || type == TYPE_BIGINT || type == TYPE_LARGEINT;
+    return type == TYPE_BOOLEAN || is_int(type);
 }
 
 constexpr bool is_decimal(PrimitiveType type) {
     return type == TYPE_DECIMAL32 || type == TYPE_DECIMAL64 || type == TYPE_DECIMAL128I ||
            type == TYPE_DECIMAL256 || type == TYPE_DECIMALV2;
 }
 
+constexpr bool is_number(PrimitiveType type) {
+    return is_int_or_bool(type) || is_float_or_double(type) || is_decimal(type);
+}
+
 PrimitiveType thrift_to_type(TPrimitiveType::type ttype);
 TPrimitiveType::type to_thrift(PrimitiveType ptype);
 std::string type_to_string(PrimitiveType t);

be/src/vec/aggregate_functions/aggregate_function_corr.cpp

@@ -39,10 +39,8 @@ AggregateFunctionPtr create_aggregate_corr_welford_function(const std::string& n
                                                             const AggregateFunctionAttr& attr) {
     assert_binary(name, argument_types);
 
-    WhichDataType which0(remove_nullable(argument_types[0]));
-    WhichDataType which1(remove_nullable(argument_types[1]));
-
-    if (!which0.is_float64() || !which1.is_float64()) {
+    if (argument_types[0]->get_primitive_type() != TYPE_DOUBLE ||
+        argument_types[1]->get_primitive_type() != TYPE_DOUBLE) {
         throw doris::Exception(ErrorCode::INTERNAL_ERROR,
                                "Aggregate function {} only support double", name);
     }

be/src/vec/aggregate_functions/aggregate_function_count_by_enum.cpp

@@ -43,9 +43,7 @@ AggregateFunctionPtr create_aggregate_function_count_by_enum(const std::string&
         type = assert_cast<const DataTypeNullable*>(type)->get_nested_type().get();
     }
 
-    WhichDataType which(*type);
-
-    if (which.is_string()) {
+    if (is_string_type(type->get_primitive_type())) {
         return std::make_shared<AggregateFunctionCountByEnum<AggregateFunctionCountByEnumData>>(
                 argument_types);
     }

be/src/vec/aggregate_functions/aggregate_function_group_array_intersect.cpp

@@ -101,7 +101,7 @@ AggregateFunctionPtr create_aggregate_function_group_array_intersect(
     assert_unary(name, argument_types);
     const DataTypePtr& argument_type = remove_nullable(argument_types[0]);
 
-    if (!WhichDataType(argument_type).is_array())
+    if (argument_type->get_primitive_type() != TYPE_ARRAY)
         throw Exception(ErrorCode::INVALID_ARGUMENT,
                         "Aggregate function groupArrayIntersect accepts only array type argument. "
                         "Provided argument type: " +

be/src/vec/aggregate_functions/aggregate_function_map.cpp

@@ -35,7 +35,6 @@ AggregateFunctionPtr create_aggregate_function_map_agg(const std::string& name,
                                                        const DataTypes& argument_types,
                                                        const bool result_is_nullable,
                                                        const AggregateFunctionAttr& attr) {
-    WhichDataType type(remove_nullable(argument_types[0]));
     switch (argument_types[0]->get_primitive_type()) {
     case PrimitiveType::TYPE_BOOLEAN:
         return create_agg_function_map_agg<UInt8>(argument_types, result_is_nullable);

be/src/vec/aggregate_functions/aggregate_function_orthogonal_bitmap.cpp

@@ -45,15 +45,13 @@ AggregateFunctionPtr create_aggregate_function_orthogonal(const std::string& nam
         return creator_without_type::create<AggFunctionOrthBitmapFunc<Impl<StringRef>>>(
                 argument_types, result_is_nullable);
     } else {
-        WhichDataType which(*remove_nullable(argument_types[1]));
-
         AggregateFunctionPtr res(
                 creator_with_type_base<true, true, false, 1>::create<AggFunctionOrthBitmapFunc,
                                                                      Impl>(argument_types,
                                                                            result_is_nullable));
         if (res) {
             return res;
-        } else if (which.is_string_or_fixed_string()) {
+        } else if (is_string_type(argument_types[1]->get_primitive_type())) {
             res = creator_without_type::create<AggFunctionOrthBitmapFunc<Impl<std::string_view>>>(
                     argument_types, result_is_nullable);
             return res;

be/src/vec/aggregate_functions/aggregate_function_rpc.h

@@ -297,22 +297,25 @@ struct AggregateRpcUdafData {
             result_type =
                     reinterpret_cast<const DataTypeNullable*>(return_type.get())->get_nested_type();
         }
-        WhichDataType which(result_type);
-        if (which.is_float32()) {
+        switch (result_type->get_primitive_type()) {
+        case TYPE_FLOAT:
             GETDATA(float, float);
-        } else if (which.is_float64()) {
+            break;
+        case TYPE_DOUBLE:
             GETDATA(double, double);
-        } else if (which.is_int32()) {
+            break;
+        case TYPE_INT:
             GETDATA(int32_t, int32);
-        } else if (which.is_uint32()) {
-            GETDATA(uint32_t, uint32);
-        } else if (which.is_int64()) {
+            break;
+        case TYPE_BIGINT:
             GETDATA(int64_t, int64);
-        } else if (which.is_uint64()) {
-            GETDATA(uint64_t, uint64);
-        } else if (which.is_uint8()) {
+            break;
+        case TYPE_BOOLEAN:
             GETDATA(uint8_t, bool);
-        } else if (which.is_string()) {
+            break;
+        case TYPE_STRING:
+        case TYPE_CHAR:
+        case TYPE_VARCHAR: {
             if (response.result_size() > 0 && response.result(0).string_value_size() > 0) {
                 std::string ret = response.result(0).string_value(0);
                 to.insert_data(ret.c_str(), ret.size());
@@ -322,7 +325,9 @@ struct AggregateRpcUdafData {
                               "result is empty";
                 to.insert_default();
             }
-        } else {
+            break;
+        }
+        default:
             LOG(ERROR) << "failed to get result cause unkown return type";
             to.insert_default();
         }