[refactor](be) Strengthen enum type safety in execution and expressions

be/benchmark/benchmark_array_range.hpp

@@ -54,12 +54,13 @@ static void BM_ArrayRange(benchmark::State& state) {
             "array_range", block.get_columns_with_type_and_name(), result_type);
     auto context =
             FunctionContext::create_context(nullptr, result_type, {int_type, int_type, int_type});
-    auto status = function->open(context.get(), FunctionContext::FRAGMENT_LOCAL);
+    auto status =
+            function->open(context.get(), FunctionContext::FunctionStateScope::FRAGMENT_LOCAL);
     if (!status.ok()) {
         state.SkipWithError(status.to_string().c_str());
         return;
     }
-    status = function->open(context.get(), FunctionContext::THREAD_LOCAL);
+    status = function->open(context.get(), FunctionContext::FunctionStateScope::THREAD_LOCAL);
     if (!status.ok()) {
         state.SkipWithError(status.to_string().c_str());
         return;
@@ -74,11 +75,11 @@ static void BM_ArrayRange(benchmark::State& state) {
         benchmark::DoNotOptimize(block.get_by_position(3).column);
         benchmark::ClobberMemory();
     }
-    status = function->close(context.get(), FunctionContext::THREAD_LOCAL);
+    status = function->close(context.get(), FunctionContext::FunctionStateScope::THREAD_LOCAL);
     if (!status.ok()) {
         state.SkipWithError(status.to_string().c_str());
     }
-    status = function->close(context.get(), FunctionContext::FRAGMENT_LOCAL);
+    status = function->close(context.get(), FunctionContext::FunctionStateScope::FRAGMENT_LOCAL);
     if (!status.ok()) {
         state.SkipWithError(status.to_string().c_str());
     }

be/benchmark/benchmark_binary_arithmetic.hpp

@@ -64,8 +64,8 @@ struct BinaryArithmeticRunner {
         }
         ctx = FunctionContext::create_context(nullptr, res_type, {type_a, type_b});
         ctx->set_check_overflow_for_decimal(true);
-        if (!func->open(ctx.get(), FunctionContext::FRAGMENT_LOCAL).ok() ||
-            !func->open(ctx.get(), FunctionContext::THREAD_LOCAL).ok()) {
+        if (!func->open(ctx.get(), FunctionContext::FunctionStateScope::FRAGMENT_LOCAL).ok() ||
+            !func->open(ctx.get(), FunctionContext::FunctionStateScope::THREAD_LOCAL).ok()) {
             throw std::runtime_error(name + " open failed");
         }
         block.insert({nullptr, res_type, "result"});

be/benchmark/benchmark_character_encoding.hpp

@@ -107,24 +107,27 @@ void BM_character_encoding(benchmark::State& state) {
         return;
     }
     auto context = FunctionContext::create_context(nullptr, result_type, {input_type, string_type});
-    auto status = function->open(context.get(), FunctionContext::FRAGMENT_LOCAL);
+    auto status =
+            function->open(context.get(), FunctionContext::FunctionStateScope::FRAGMENT_LOCAL);
     if (!status.ok()) {
         state.SkipWithError(status.to_string());
         return;
     }
     Defer close_fragment {[&] {
-        auto close_status = function->close(context.get(), FunctionContext::FRAGMENT_LOCAL);
+        auto close_status =
+                function->close(context.get(), FunctionContext::FunctionStateScope::FRAGMENT_LOCAL);
         if (!close_status.ok()) {
             state.SkipWithError(close_status.to_string());
         }
     }};
-    status = function->open(context.get(), FunctionContext::THREAD_LOCAL);
+    status = function->open(context.get(), FunctionContext::FunctionStateScope::THREAD_LOCAL);
     if (!status.ok()) {
         state.SkipWithError(status.to_string());
         return;
     }
     Defer close_thread {[&] {
-        auto close_status = function->close(context.get(), FunctionContext::THREAD_LOCAL);
+        auto close_status =
+                function->close(context.get(), FunctionContext::FunctionStateScope::THREAD_LOCAL);
         if (!close_status.ok()) {
             state.SkipWithError(close_status.to_string());
         }

be/benchmark/variant/benchmark_variant_segment.hpp

@@ -369,11 +369,15 @@ Status parse_json_batch(ColumnPtr json, const LayoutConfig& config, uint32_t row
     DORIS_CHECK(function != nullptr);
     block.insert({nullptr, result_type, "result"});
     auto context = FunctionContext::create_context(nullptr, result_type, {input_type});
-    RETURN_IF_ERROR(function->open(context.get(), FunctionContext::FRAGMENT_LOCAL));
-    RETURN_IF_ERROR(function->open(context.get(), FunctionContext::THREAD_LOCAL));
+    RETURN_IF_ERROR(
+            function->open(context.get(), FunctionContext::FunctionStateScope::FRAGMENT_LOCAL));
+    RETURN_IF_ERROR(
+            function->open(context.get(), FunctionContext::FunctionStateScope::THREAD_LOCAL));
     Status status = function->execute(context.get(), block, {0}, 1, rows);
-    Status close_thread = function->close(context.get(), FunctionContext::THREAD_LOCAL);
-    Status close_fragment = function->close(context.get(), FunctionContext::FRAGMENT_LOCAL);
+    Status close_thread =
+            function->close(context.get(), FunctionContext::FunctionStateScope::THREAD_LOCAL);
+    Status close_fragment =
+            function->close(context.get(), FunctionContext::FunctionStateScope::FRAGMENT_LOCAL);
     RETURN_IF_ERROR(status);
     RETURN_IF_ERROR(close_thread);
     RETURN_IF_ERROR(close_fragment);

be/src/core/column/column_vector.h

@@ -20,6 +20,7 @@
 
 #pragma once
 
+#include <gen_cpp/PlanNodes_types.h>
 #include <glog/logging.h>
 #include <sys/types.h>
 
@@ -54,7 +55,7 @@
 class SipHash;
 
 namespace doris {
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 struct ProcessHashTableProbe;
 }
 
@@ -77,7 +78,7 @@ class ColumnVector final : public COWHelper<IColumn, ColumnVector<T>> {
     using Self = ColumnVector;
     friend class COWHelper<IColumn, Self>;
 
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     friend struct doris::ProcessHashTableProbe;
 
     struct less;

be/src/exec/common/hash_table/join_hash_table.h

@@ -47,7 +47,7 @@ class JoinHashTable {
         return cal_vector_mem(visited) + cal_vector_mem(first) + cal_vector_mem(next);
     }
 
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     void prepare_build(size_t num_elem, int batch_size, bool has_null_key,
                        uint32_t force_bucket_size) {
         _has_null_key = has_null_key;
@@ -99,7 +99,7 @@ class JoinHashTable {
         _keep_null_key = keep_null_key;
     }
 
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     auto find_batch(const Key* __restrict keys, const uint32_t* __restrict build_idx_map,
                     int probe_idx, uint32_t build_idx, int probe_rows,
                     uint32_t* __restrict probe_idxs, bool& probe_visited,
@@ -196,7 +196,7 @@ class JoinHashTable {
         }
     }
 
-    template <int JoinOpType, bool is_mark_join>
+    template <TJoinOp::type JoinOpType, bool is_mark_join>
     bool iterate_map(ColumnOffset32& build_idxs, ColumnFilterHelper* mark_column_helper) const {
         const auto batch_size = max_batch_size;
         const auto elem_num = visited.size();
@@ -241,7 +241,7 @@ class JoinHashTable {
         return lhs == rhs;
     }
 
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     auto _process_null_aware_left_half_join_for_empty_build_side(int probe_idx, int probe_rows,
                                                                  uint32_t* __restrict probe_idxs,
                                                                  uint32_t* __restrict build_idxs) {
@@ -289,7 +289,7 @@ class JoinHashTable {
         return std::tuple {probe_idx, 0U, 0U};
     }
 
-    template <int JoinOpType, bool has_null_map>
+    template <TJoinOp::type JoinOpType, bool has_null_map>
     auto _find_batch_left_semi_anti(const Key* __restrict keys,
                                     const uint32_t* __restrict build_idx_map, int probe_idx,
                                     int probe_rows, uint32_t* __restrict probe_idxs,
@@ -317,7 +317,7 @@ class JoinHashTable {
         return std::tuple {probe_idx, 0U, matched_cnt};
     }
 
-    template <int JoinOpType, bool only_need_to_match_one>
+    template <TJoinOp::type JoinOpType, bool only_need_to_match_one>
     auto _find_batch_conjunct(const Key* __restrict keys, const uint32_t* __restrict build_idx_map,
                               int probe_idx, uint32_t build_idx, int probe_rows,
                               uint32_t* __restrict probe_idxs, uint32_t* __restrict build_idxs) {
@@ -370,7 +370,7 @@ class JoinHashTable {
         return std::tuple {probe_idx, build_idx, matched_cnt};
     }
 
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     auto _find_batch_inner_outer_join(const Key* __restrict keys,
                                       const uint32_t* __restrict build_idx_map, int probe_idx,
                                       uint32_t build_idx, int probe_rows,

be/src/exec/common/join_op_utils.h

@@ -52,11 +52,11 @@ inline bool is_asof_join(TJoinOp::type join_op) {
     return join_op == TJoinOp::ASOF_LEFT_INNER_JOIN || join_op == TJoinOp::ASOF_LEFT_OUTER_JOIN;
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 inline constexpr bool is_asof_join_op_v =
         JoinOpType == TJoinOp::ASOF_LEFT_INNER_JOIN || JoinOpType == TJoinOp::ASOF_LEFT_OUTER_JOIN;
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 inline constexpr bool is_asof_outer_join_op_v = JoinOpType == TJoinOp::ASOF_LEFT_OUTER_JOIN;
 
 // ASOF JOIN index with inline values for cache-friendly branchless binary search.

be/src/exec/operator/hashjoin_build_sink.h

@@ -216,7 +216,7 @@ struct ProcessHashTableBuild {
               _batch_size(batch_size),
               _state(state) {}
 
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     Status run(HashTableContext& hash_table_ctx, ConstNullMapPtr null_map, bool* has_null_key,
                bool short_circuit_for_null, bool with_other_conjuncts) {
         if (null_map) {

be/src/exec/operator/hashjoin_probe_operator.h

@@ -73,7 +73,7 @@ class HashJoinProbeLocalState MOCK_REMOVE(final)
     std::vector<uint16_t> _convert_block_to_null(Block& block);
     Status _extract_join_column(Block& block, const std::vector<int>& res_col_ids);
     friend class HashJoinProbeOperatorX;
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     friend struct ProcessHashTableProbe;
 
     // one probe row may match multiple build rows. and we need output whenever we filled a block.
@@ -184,7 +184,7 @@ class HashJoinProbeOperatorX MOCK_REMOVE(final)
                         RuntimeProfile::Counter& expr_call_timer,
                         std::vector<int>& res_col_ids) const;
     friend class HashJoinProbeLocalState;
-    template <int JoinOpType>
+    template <TJoinOp::type JoinOpType>
     friend struct ProcessHashTableProbe;
 
     const TJoinDistributionType::type _join_distribution;

be/src/exec/operator/join/process_hash_table_probe.h

@@ -39,7 +39,7 @@ using MutableColumns = std::vector<MutableColumnPtr>;
 using NullMap = ColumnUInt8::Container;
 using ConstNullMapPtr = const NullMap*;
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 struct ProcessHashTableProbe {
     ProcessHashTableProbe(HashJoinProbeLocalState* parent, int batch_size);
     ~ProcessHashTableProbe() = default;

be/src/exec/operator/join/process_hash_table_probe_impl.h

@@ -66,7 +66,7 @@ static void mock_column_size(auto& col, size_t size) {
     }
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 ProcessHashTableProbe<JoinOpType>::ProcessHashTableProbe(HashJoinProbeLocalState* parent,
                                                          int batch_size)
         : _parent(parent),
@@ -97,7 +97,7 @@ ProcessHashTableProbe<JoinOpType>::ProcessHashTableProbe(HashJoinProbeLocalState
             probe_output_non_lazy_materialized_count >= CONTIGUOUS_INDICES_CHECK_THRESHOLD;
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 void ProcessHashTableProbe<JoinOpType>::build_side_output_column(MutableColumns& mcol,
                                                                  bool is_mark_join) {
     SCOPED_TIMER(_build_side_output_timer);
@@ -156,7 +156,7 @@ void ProcessHashTableProbe<JoinOpType>::build_side_output_column(MutableColumns&
     }
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 bool ProcessHashTableProbe<JoinOpType>::can_transfer_probe_columns_to_output(
         bool probe_indices_are_contiguous) const {
     constexpr bool probe_side_type_unchanged =
@@ -175,7 +175,7 @@ bool ProcessHashTableProbe<JoinOpType>::can_transfer_probe_columns_to_output(
            _parent->_probe_index == _parent->_probe_block.rows() && _parent->_build_index == 0;
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 void ProcessHashTableProbe<JoinOpType>::probe_side_output_column(MutableColumns& mcol) {
     SCOPED_TIMER(_probe_side_output_timer);
     auto& probe_block = _parent->_probe_block;
@@ -209,7 +209,7 @@ void ProcessHashTableProbe<JoinOpType>::probe_side_output_column(MutableColumns&
     }
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 template <typename HashTableType>
 typename HashTableType::State ProcessHashTableProbe<JoinOpType>::_init_probe_side(
         HashTableType& hash_table_ctx, uint32_t probe_rows, const uint8_t* null_map) {
@@ -249,7 +249,7 @@ typename HashTableType::State ProcessHashTableProbe<JoinOpType>::_init_probe_sid
 // Phase 2: Batch binary search — find ASOF best match for each resolved row.
 // This separation improves cache locality since phase 2 accesses AsofIndexGroup
 // data sequentially without interleaving hash chain walks.
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 template <typename HashTableType>
 uint32_t ProcessHashTableProbe<JoinOpType>::
         _find_batch_asof_optimized( // NOLINT(readability-function-cognitive-complexity)
@@ -507,7 +507,7 @@ uint32_t ProcessHashTableProbe<JoinOpType>::
     return matched_cnt;
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 template <typename HashTableType>
 void ProcessHashTableProbe<JoinOpType>::process_direct_return(HashTableType& hash_table_ctx,
                                                               MutableBlock& mutable_block,
@@ -524,7 +524,7 @@ void ProcessHashTableProbe<JoinOpType>::process_direct_return(HashTableType& has
     output_block->swap(mutable_block.to_block());
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 template <typename HashTableType>
 Status ProcessHashTableProbe<JoinOpType>::process(HashTableType& hash_table_ctx,
                                                   const uint8_t* null_map,
@@ -645,7 +645,7 @@ Status ProcessHashTableProbe<JoinOpType>::process(HashTableType& hash_table_ctx,
     return Status::OK();
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 uint32_t ProcessHashTableProbe<JoinOpType>::_process_probe_null_key(uint32_t probe_index) {
     const auto rows = _build_block->rows();
 
@@ -669,7 +669,7 @@ uint32_t ProcessHashTableProbe<JoinOpType>::_process_probe_null_key(uint32_t pro
     return matched_cnt;
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 Status ProcessHashTableProbe<JoinOpType>::finalize_block_with_filter(Block* output_block,
                                                                      size_t filter_column_id,
                                                                      size_t column_to_keep) {
@@ -755,7 +755,7 @@ Status ProcessHashTableProbe<JoinOpType>::finalize_block_with_filter(Block* outp
      * To avoid using nested loop join, we use the mark join conjunct(`t1.k1 = t2.k1`) as the equal join conjunct.
      * So this query will be a "null aware left anti join", which means the equal conjunct's result should be nullable.
      */
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 Status ProcessHashTableProbe<JoinOpType>::do_mark_join_conjuncts(Block* output_block,
                                                                  const uint8_t* null_map) {
     if (JoinOpType != TJoinOp::LEFT_ANTI_JOIN && JoinOpType != TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN &&
@@ -909,7 +909,7 @@ Status ProcessHashTableProbe<JoinOpType>::do_mark_join_conjuncts(Block* output_b
     }
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 Status ProcessHashTableProbe<JoinOpType>::do_other_join_conjuncts(Block* output_block,
                                                                   DorisVector<uint8_t>& visited) {
     // dispose the other join conjunct exec
@@ -1025,7 +1025,7 @@ Status ProcessHashTableProbe<JoinOpType>::do_other_join_conjuncts(Block* output_
     return Status::OK();
 }
 
-template <int JoinOpType>
+template <TJoinOp::type JoinOpType>
 template <typename HashTableType>
 Status ProcessHashTableProbe<
         JoinOpType>::finish_probing( // NOLINT(readability-function-cognitive-complexity)

be/src/exec/operator/scan_operator.cpp

@@ -744,7 +744,7 @@ Status ScanLocalStateBase::_normalize_in_predicate(
         // begin to push InPredicate value into ColumnValueRange
         auto* state = reinterpret_cast<InState*>(
                 expr_ctx->fn_context(tmp->fn_context_index())
-                        ->get_function_state(FunctionContext::FRAGMENT_LOCAL));
+                        ->get_function_state(FunctionContext::FunctionStateScope::FRAGMENT_LOCAL));
 
         // xx in (col, xx, xx) should not be push down
         if (!state->use_set) {

be/src/exec/runtime_filter/utils.cpp

@@ -75,7 +75,7 @@ RuntimeFilterType get_runtime_filter_type(const TRuntimeFilterDesc* desc) {
 }
 
 // PFilterType -> RuntimeFilterType
-RuntimeFilterType get_type(int filter_type) {
+RuntimeFilterType get_type(PFilterType filter_type) {
     switch (filter_type) {
     case PFilterType::IN_FILTER: {
         return RuntimeFilterType::IN_FILTER;

be/src/exec/runtime_filter/utils.h

@@ -90,7 +90,7 @@ std::string filter_type_to_string(RuntimeFilterType type);
 RuntimeFilterType get_runtime_filter_type(const TRuntimeFilterDesc* desc);
 
 // PFilterType -> RuntimeFilterType
-RuntimeFilterType get_type(int filter_type);
+RuntimeFilterType get_type(PFilterType filter_type);
 // RuntimeFilterType -> PFilterType
 PFilterType get_type(RuntimeFilterType type);
 

be/src/exec/sink/vtablet_finder.h

@@ -39,7 +39,7 @@ class OlapTabletFinder {
     // which indicates that we should only compute tablet index in the corresponding partition once for the
     // whole time in olap table sink
     // FIND_TABLET_RANDOM_BUCKET is used for V1 adaptive random bucket mode.
-    enum FindTabletMode {
+    enum class FindTabletMode {
         FIND_TABLET_EVERY_ROW,
         FIND_TABLET_EVERY_BATCH,
         FIND_TABLET_EVERY_SINK,