[Refactor] Extract exec join core module (StarRocks#72303)

Signed-off-by: alvin-celerdata <alvin.zhao@celerdata.com>

Author: alvin-phoenix-ai

be/AGENTS.md

@@ -205,6 +205,14 @@ Reusable exec sorting algorithms without pipeline, spill, workgroup, storage, or
 - Allowed target deps: `ExprCore`, `RuntimeCore`, `ChunkCore`, `ColumnCore`, `TypesCore`, `Common`, `Base`, `Gutil`, `StarRocksGen`
 - Core tests: `exec_sorting_core_test`
 - Remediation: Keep ExecSortingCore limited to reusable sorting algorithms; leave merge-path pipeline observer integration, chunk sorters, spill-aware sorting, and exec-node adapters in higher modules.
+
+### ExecJoinCore (`execjoincore`)
+Reusable exec join hash table algorithms without join nodes, pipeline, storage, service, or util coupling.
+- Targets: `ExecJoinCore`
+- Allowed internal include prefixes: `exec/join/`, `exec/sorting/sort_helper.h`, `exprs/`, `serde/`, `runtime/`, `column/`, `types/`, `common/`, `base/`, `gutil/`, `gen_cpp/`
+- Allowed target deps: `ExecSortingCore`, `ExprCore`, `SerdeCore`, `RuntimeCore`, `ChunkCore`, `ColumnCore`, `TypesCore`, `Common`, `Base`, `Gutil`, `StarRocksGen`
+- Core tests: `exec_join_core_test`
+- Remediation: Keep ExecJoinCore limited to reusable join hash table algorithms; leave join nodes, pipeline operators, storage/service integration, and util diagnostics in higher modules.
 <!-- END GENERATED: BE MODULE HARNESSES -->
 
 ## BE-Specific Sync Rules

be/module_boundary_manifest.json

@@ -421,6 +421,30 @@
       "allowed_test_targets": ["exec_sorting_core_test"],
       "allowed_test_link_deps": ["ExecSortingCore", "ExprCore", "RuntimeCore", "ChunkCore", "ColumnCore", "TypesCore", "Common", "Base", "Gutil", "StarRocksGen"],
       "remediation": "Keep ExecSortingCore limited to reusable sorting algorithms; leave merge-path pipeline observer integration, chunk sorters, spill-aware sorting, and exec-node adapters in higher modules."
+    },
+    {
+      "id": "execjoincore",
+      "doc_label": "ExecJoinCore",
+      "summary": "Reusable exec join hash table algorithms without join nodes, pipeline, storage, service, or util coupling.",
+      "owned_targets": ["ExecJoinCore"],
+      "owned_globs": [
+        "be/src/exec/join/**"
+      ],
+      "allowed_include_prefixes": ["exec/join/", "exec/sorting/sort_helper.h", "exprs/", "serde/", "runtime/", "column/", "types/", "common/", "base/", "gutil/", "gen_cpp/"],
+      "forbidden_include_prefixes": ["exec/", "exec/pipeline/", "exec/spill/", "exec/workgroup/", "runtime/batch_write/", "runtime/routine_load/", "runtime/stream_load/", "storage/", "service/", "http/", "cache/", "connector/", "util/"],
+      "forbidden_includes": [
+        "exec/cross_join_node.h",
+        "exec/data_sink.h",
+        "exec/exec_node.h",
+        "exec/hash_join_components.h",
+        "exec/hash_join_node.h",
+        "exec/hash_joiner.h",
+        "runtime/exec_env.h"
+      ],
+      "allowed_target_deps": ["ExecSortingCore", "ExprCore", "SerdeCore", "RuntimeCore", "ChunkCore", "ColumnCore", "TypesCore", "Common", "Base", "Gutil", "StarRocksGen"],
+      "allowed_test_targets": ["exec_join_core_test"],
+      "allowed_test_link_deps": ["ExecJoinCore", "ExecSortingCore", "ExprCore", "SerdeCore", "RuntimeCore", "ChunkCore", "ColumnCore", "TypesCore", "Common", "Base", "Gutil", "StarRocksGen"],
+      "remediation": "Keep ExecJoinCore limited to reusable join hash table algorithms; leave join nodes, pipeline operators, storage/service integration, and util diagnostics in higher modules."
     }
   ]
 }

be/src/column/CMakeLists.txt

@@ -67,4 +67,4 @@ ADD_BE_LIB(Column
 )
 
 target_link_libraries(ChunkCore PUBLIC ColumnCore)
-target_link_libraries(Column PUBLIC ColumnCore ChunkCore)
+target_link_libraries(Column PUBLIC ColumnCore ChunkCore SerdeCore)

be/src/exec/AGENTS.md

@@ -13,8 +13,8 @@ Local guide for work under `be/src/exec`.
 ## Current Direction
 
 - Keep `Exec` as the temporary compatibility umbrella while extracted modules move out of it.
-- Treat existing targets such as `ExecCore`, `ExecSinkCore`, `SpillCore`, `ExecSchemaScannerCore`, `ExecSchemaScanners`, and `ExecSortingCore` as the model for future extractions.
-- Follow `handbook/plans/active/2026-04-28-exec-layering-rollout.md` for the active exec module roadmap.
+- Treat existing targets such as `ExecCore`, `ExecSinkCore`, `SpillCore`, `ExecSchemaScannerCore`, `ExecSchemaScanners`, `ExecSortingCore`, and `ExecJoinCore` as the model for future extractions.
+- Keep local or checkout-specific roadmap notes outside the committed guide unless they become shared repository policy.
 
 ## Validation
 

be/src/exec/CMakeLists.txt

@@ -135,6 +135,44 @@ target_link_libraries(ExecSortingCore PUBLIC
     StarRocksGen
 )
 
+set(EXEC_JOIN_CORE_FILES
+    join/join_hash_table_descriptor.cpp
+    join/join_key_constructor.cpp
+    join/join_hash_table.cpp
+    join/join_hash_map_mapping_inst.cpp
+    join/join_hash_map_linear_chained_inst.cpp
+    join/join_hash_map_linear_chained_set_inst.cpp
+    join/join_hash_map_bucket_chained_inst.cpp
+    join/join_hash_map_asof_inst.cpp
+)
+
+ADD_BE_LIB(ExecJoinCore
+    ${EXEC_JOIN_CORE_FILES}
+)
+
+target_link_libraries(ExecJoinCore PUBLIC
+    ExecSortingCore
+    ExprCore
+    SerdeCore
+    RuntimeCore
+    ChunkCore
+    ColumnCore
+    TypesCore
+    Common
+    Base
+    Gutil
+    StarRocksGen
+)
+
+if (WITH_PCH)
+    target_precompile_headers(ExecJoinCore
+        PRIVATE
+        join/join_hash_map.h
+        join/join_hash_map.hpp
+        join/join_hash_map_method.hpp
+    )
+endif()
+
 # Staged pipeline compatibility groups. These source lists are not final clean
 # module boundaries yet; keep them named by intended ownership so later
 # cycle-cut leaves can promote them into real targets incrementally.
@@ -361,9 +399,6 @@ set(EXEC_NON_PIPELINE_FILES
     hash_joiner.cpp
     hash_join_node.cpp
     hash_join_components.cpp
-    join/join_hash_table_descriptor.cpp
-    join/join_key_constructor.cpp
-    join/join_hash_table.cpp
     topn_node.cpp
     chunks_sorter.cpp
     chunks_sorter_heap_sort.cpp
@@ -494,38 +529,19 @@ add_library(pipeline_extensions_exec_objs OBJECT
     ${PIPELINE_EXTENSIONS_SRCS}
 )
 
-add_library(join_hash_map_inst_objs OBJECT
-    join/join_hash_map_mapping_inst.cpp
-    join/join_hash_map_linear_chained_inst.cpp
-    join/join_hash_map_linear_chained_set_inst.cpp
-    join/join_hash_map_bucket_chained_inst.cpp
-    join/join_hash_map_asof_inst.cpp
-)
-
 IF(ENABLE_MULTI_DYNAMIC_LIBS)
     set_property(TARGET pipeline_primitives_exec_objs PROPERTY POSITION_INDEPENDENT_CODE ON)
     set_property(TARGET pipeline_runtime_compat_exec_objs PROPERTY POSITION_INDEPENDENT_CODE ON)
     set_property(TARGET pipeline_operators_batch_exec_objs PROPERTY POSITION_INDEPENDENT_CODE ON)
     set_property(TARGET pipeline_extensions_exec_objs PROPERTY POSITION_INDEPENDENT_CODE ON)
-    set_property(TARGET join_hash_map_inst_objs PROPERTY POSITION_INDEPENDENT_CODE ON)
 ENDIF()
 
-if (WITH_PCH)
-    target_precompile_headers(join_hash_map_inst_objs
-        PRIVATE
-        join/join_hash_map.h
-        join/join_hash_map.hpp
-        join/join_hash_map_method.hpp
-    )
-endif()
-
 ADD_BE_LIB(Exec
     ${EXEC_NON_PIPELINE_FILES}
     $<TARGET_OBJECTS:pipeline_primitives_exec_objs>
     $<TARGET_OBJECTS:pipeline_runtime_compat_exec_objs>
     $<TARGET_OBJECTS:pipeline_operators_batch_exec_objs>
     $<TARGET_OBJECTS:pipeline_extensions_exec_objs>
-    $<TARGET_OBJECTS:join_hash_map_inst_objs>
 )
 
-target_link_libraries(Exec PUBLIC ExecSinkCore ExecCore SpillCore ExecSchemaScannerCore ExecSchemaScanners ExecSortingCore)
+target_link_libraries(Exec PUBLIC ExecSinkCore ExecCore SpillCore ExecSchemaScannerCore ExecSchemaScanners ExecSortingCore ExecJoinCore)

be/src/exec/join/join_hash_table.cpp

@@ -22,15 +22,14 @@
 #include "column/chunk.h"
 #include "column/vectorized_fwd.h"
 #include "common/runtime_profile.h"
+#include "common/stack_util.h"
 #include "common/statusor.h"
 #include "common/system/cpu_info.h"
-#include "exec/hash_join_node.h"
 #include "exec/join/join_hash_map_method.h"
 #include "exec/join/join_key_constructor.h"
+#include "exprs/column_ref.h"
 #include "runtime/descriptors.h"
-#include "serde/column_array_serde.h"
 #include "types/logical_type_infra.h"
-#include "util/stack_util.h"
 
 namespace starrocks {
 

be/src/serde/CMakeLists.txt

@@ -17,9 +17,34 @@ set(LIBRARY_OUTPUT_PATH "${BUILD_DIR}/src/serde")
 # where to put generated binaries
 set(EXECUTABLE_OUTPUT_PATH "${BUILD_DIR}/src/serde")
 
+ADD_BE_LIB(SerdeCore
+        column_array_serde.cpp
+        )
+
+target_link_libraries(SerdeCore PUBLIC
+        ColumnCore
+        TypesCore
+        Common
+        Base
+        Gutil
+        StarRocksGen
+        )
+
 ADD_BE_LIB(Serde
         encode_context.cpp
-        column_array_serde.cpp
         compress_strategy.cpp
         protobuf_serde.cpp
         )
+
+target_link_libraries(Serde PUBLIC
+        SerdeCore
+        Column
+        RuntimeCore
+        ChunkCore
+        ColumnCore
+        TypesCore
+        Common
+        Base
+        Gutil
+        StarRocksGen
+        )

be/src/serde/column_array_serde.cpp

@@ -39,7 +39,7 @@
 #include "column/variant_column.h"
 #include "column/vectorized_fwd.h"
 #include "common/config_diagnostic_fwd.h"
-#include "serde/protobuf_serde.h"
+#include "serde/encode_level.h"
 #include "types/json_value.h"
 #include "types/percentile_value.h"
 
@@ -214,7 +214,7 @@ class FixedLengthColumnSerde {
         // NOTE that `serialize` and `deserialize` will store and load the size as uint32_t.
         // If you use `serialize` and `deserialize`, please make sure that the size of the column is less than 2^32.
         int64_t size = sizeof(T) * column.size();
-        if (EncodeContext::enable_encode_integer(encode_level) && size >= ENCODE_SIZE_LIMIT) {
+        if (is_integer_encoding_enabled(encode_level) && size >= ENCODE_SIZE_LIMIT) {
             return sizeof(uint32_t) + sizeof(uint64_t) +
                    std::max((int64_t)size, (int64_t)streamvbyte_max_compressedbytes(upper_int32(size)));
         } else {
@@ -225,7 +225,7 @@ class FixedLengthColumnSerde {
     static uint8_t* serialize(const FixedLengthColumnBase<T>& column, uint8_t* buff, const int encode_level) {
         uint32_t size = sizeof(T) * column.size();
         buff = write_little_endian_32(size, buff);
-        if (EncodeContext::enable_encode_integer(encode_level) && size >= ENCODE_SIZE_LIMIT) {
+        if (is_integer_encoding_enabled(encode_level) && size >= ENCODE_SIZE_LIMIT) {
             // sorted 32-bit integers have a better optimize branch
             buff = encode_integers<(sizeof(T) == 4 && sorted)>(column.raw_data(), size, buff, encode_level);
         } else {
@@ -240,7 +240,7 @@ class FixedLengthColumnSerde {
         ASSIGN_OR_RETURN(buff, read_little_endian_32(buff, end, &size));
         auto& data = column->get_data();
         raw::make_room(&data, size / sizeof(T));
-        if (EncodeContext::enable_encode_integer(encode_level) && size >= ENCODE_SIZE_LIMIT) {
+        if (is_integer_encoding_enabled(encode_level) && size >= ENCODE_SIZE_LIMIT) {
             constexpr bool is_sorted_i32 = sizeof(T) == 4 && sorted;
             ASSIGN_OR_RETURN(buff, decode_integers<is_sorted_i32>(buff, end, data.data(), size));
         } else {
@@ -258,13 +258,13 @@ class BinaryColumnSerde {
         const auto& offsets = column.get_offset();
         int64_t res = sizeof(T) * 2;
         int64_t offsets_size = offsets.size() * sizeof(typename BinaryColumnBase<T>::Offset);
-        if (EncodeContext::enable_encode_integer(encode_level) && offsets_size >= ENCODE_SIZE_LIMIT) {
+        if (is_integer_encoding_enabled(encode_level) && offsets_size >= ENCODE_SIZE_LIMIT) {
             res += sizeof(uint64_t) +
                    std::max((int64_t)offsets_size, (int64_t)streamvbyte_max_compressedbytes(upper_int32(offsets_size)));
         } else {
             res += offsets_size;
         }
-        if (EncodeContext::enable_encode_string(encode_level) && bytes.size() >= ENCODE_SIZE_LIMIT) {
+        if (is_string_encoding_enabled(encode_level) && bytes.size() >= ENCODE_SIZE_LIMIT) {
             res += sizeof(uint64_t) + std::max((int64_t)bytes.size(), (int64_t)LZ4_compressBound(bytes.size()));
         } else {
             res += bytes.size();
@@ -283,7 +283,7 @@ class BinaryColumnSerde {
         } else {
             buff = write_little_endian_64(bytes_size, buff);
         }
-        if (EncodeContext::enable_encode_string(encode_level) && bytes_size >= ENCODE_SIZE_LIMIT &&
+        if (is_string_encoding_enabled(encode_level) && bytes_size >= ENCODE_SIZE_LIMIT &&
             bytes_size <= LZ4_MAX_INPUT_SIZE) {
             buff = encode_string_lz4(bytes.data(), bytes_size, buff, encode_level);
         } else {
@@ -297,7 +297,7 @@ class BinaryColumnSerde {
         } else {
             buff = write_little_endian_64(offsets_size, buff);
         }
-        if (EncodeContext::enable_encode_integer(encode_level) && offsets_size >= ENCODE_SIZE_LIMIT) {
+        if (is_integer_encoding_enabled(encode_level) && offsets_size >= ENCODE_SIZE_LIMIT) {
             if (sizeof(T) == 4) { // only support sorted 32-bit integers
                 buff = encode_integers<true>(offsets.data(), offsets_size, buff, encode_level);
             } else {
@@ -322,7 +322,7 @@ class BinaryColumnSerde {
         column->get_bytes().resize(bytes_size);
 
         auto* bytes_data = column->get_bytes().data();
-        if (EncodeContext::enable_encode_string(encode_level) && bytes_size >= ENCODE_SIZE_LIMIT &&
+        if (is_string_encoding_enabled(encode_level) && bytes_size >= ENCODE_SIZE_LIMIT &&
             bytes_size <= LZ4_MAX_INPUT_SIZE) {
             ASSIGN_OR_RETURN(buff, decode_string_lz4(buff, end, bytes_data, bytes_size));
         } else {
@@ -337,7 +337,7 @@ class BinaryColumnSerde {
         }
         raw::make_room(&column->get_offset(), offset_bytes_size / sizeof(typename BinaryColumnBase<T>::Offset));
 
-        if (EncodeContext::enable_encode_integer(encode_level) && offset_bytes_size >= ENCODE_SIZE_LIMIT) {
+        if (is_integer_encoding_enabled(encode_level) && offset_bytes_size >= ENCODE_SIZE_LIMIT) {
             constexpr bool is_i32 = sizeof(T) == 4;
             ASSIGN_OR_RETURN(buff, decode_integers<is_i32>(buff, end, column->get_offset().data(), offset_bytes_size));
         } else {

be/src/serde/encode_context.h

@@ -22,6 +22,7 @@
 #include "column/chunk.h"
 #include "column/chunk_extra_data.h"
 #include "common/statusor.h"
+#include "serde/encode_level.h"
 
 namespace starrocks {
 class ChunkPB;
@@ -60,14 +61,11 @@ class EncodeContext {
 
     static constexpr uint16_t STREAMVBYTE_PADDING_SIZE = STREAMVBYTE_PADDING;
 
-    static bool enable_encode_integer(const int encode_level) { return encode_level & ENCODE_INTEGER; }
+    static bool enable_encode_integer(const int encode_level) { return is_integer_encoding_enabled(encode_level); }
 
-    static bool enable_encode_string(const int encode_level) { return encode_level & ENCODE_STRING; }
+    static bool enable_encode_string(const int encode_level) { return is_string_encoding_enabled(encode_level); }
 
 private:
-    static constexpr int ENCODE_INTEGER = 2;
-    static constexpr int ENCODE_STRING = 4;
-
     // if encode ratio < EncodeRatioLimit, encode it, otherwise not.
     void _adjust(const int col_id);
     const int _session_encode_level;
@@ -77,4 +75,4 @@ class EncodeContext {
     std::vector<uint64_t> _raw_bytes, _encoded_bytes;
     std::vector<uint32_t> _column_encode_level;
 };
-} // namespace starrocks::serde
+} // namespace starrocks::serde

be/src/serde/encode_level.h

@@ -0,0 +1,30 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+namespace starrocks::serde {
+
+constexpr int ENCODE_INTEGER = 2;
+constexpr int ENCODE_STRING = 4;
+
+inline bool is_integer_encoding_enabled(const int encode_level) {
+    return encode_level & ENCODE_INTEGER;
+}
+
+inline bool is_string_encoding_enabled(const int encode_level) {
+    return encode_level & ENCODE_STRING;
+}
+
+} // namespace starrocks::serde