Merge remote-tracking branch 'base/main' into sdcg-poc

# Conflicts:
#	be/src/exec/file_scanner/json_scanner.cpp

Author: meegoo

be/AGENTS.md

@@ -174,13 +174,13 @@ Core runtime building blocks without full Runtime/Exec/Storage coupling.
 - Core tests: `runtime_core_test`
 - Remediation: Keep RuntimeCore restricted to core runtime infrastructure; move service/storage/stream-load/integration code into Runtime.
 
-### FormatsIO (`formatsio`)
-Format-oriented output stream primitives above RuntimeCore, FSCore, and Types.
-- Targets: `FormatsIO`
-- Allowed internal include prefixes: `formats/io/`, `runtime/`, `fs/`, `types/`, `common/`, `base/`, `gutil/`, `gen_cpp/`
-- Allowed target deps: `RuntimeCore`, `FSCore`, `Types`, `Common`, `Base`, `Gutil`, `StarRocksGen`
-- Core tests: `formats_io_test`
-- Remediation: Keep FormatsIO limited to reusable format output stream primitives; move connector orchestration and higher execution policy upward.
+### FormatCore (`formatcore`)
+Format-oriented core primitives above ComputeEnv, ExprCore, RuntimeCore, FSCore, ChunkCore, ColumnCore, and Types.
+- Targets: `FormatCore`
+- Allowed internal include prefixes: `formats/column_evaluator.h`, `formats/deletion_bitmap.h`, `formats/disk_range.hpp`, `formats/file_writer.h`, `formats/io/`, `formats/utils.h`, `exprs/`, `runtime/`, `fs/`, `column/`, `types/`, `common/`, `base/`, `gutil/`, `gen_cpp/`
+- Allowed target deps: `ComputeEnv`, `ExprCore`, `RuntimeCore`, `FSCore`, `ChunkCore`, `ColumnCore`, `Types`, `Common`, `Base`, `Gutil`, `StarRocksGen`
+- Core tests: `format_test`
+- Remediation: Keep FormatCore limited to reusable format primitives; move connector orchestration and higher execution policy upward.
 
 ### RuntimeEnv (`runtimeenv`)
 Process-scoped runtime environment resources below full Runtime and above RuntimeCore.
@@ -207,12 +207,12 @@ Base storage algorithms and mask-buffer helpers above StoragePrimitive and Compu
 - Remediation: Keep StorageBase limited to reusable storage algorithms and helpers that may depend on ComputeEnv; move concrete storage engine, tablet, rowset, lake, service, connector, cache, and full Exec integration upward.
 
 ### ComputeEnv (`computeenv`)
-Shared compute-side BE environment boundary for process-scoped compute resources, query-scoped scan coordination helpers, query-cache primitives, WorkGroup scheduling/executor resources, and pipeline controls below full Exec/Storage and above RuntimeEnv.
+Shared compute-side BE environment boundary for process-scoped compute resources, load-path management, query-scoped scan coordination helpers, query-cache primitives, WorkGroup scheduling/executor resources, and pipeline controls below full Exec/Storage and above RuntimeEnv.
 - Targets: `ComputeEnv`
 - Allowed internal include prefixes: `compute_env/`, `exec/runtime_filter/`, `exec/pipeline/pipeline_fwd.h`, `exec/pipeline/operator.h`, `exec/pipeline/primitives/`, `exec/pipeline/runtime_filter_core_types.h`, `exec/pipeline/scan/scan_morsel.h`, `exec/pipeline/scan/morsel_queue.h`, `exec/pipeline/scan/morsel_queue_builder.h`, `exec/pipeline/scan/fixed_morsel_queue.h`, `exec/pipeline/scan/fixed_morsel_queue_builder.h`, `exec/pipeline/scan/dynamic_morsel_queue.h`, `exec/pipeline/scan/dynamic_morsel_queue_builder.h`, `exec/pipeline/scan/ticketed_morsel_queue.h`, `storage/primitive/`, `exprs/`, `serde/`, `runtime/env/`, `runtime/`, `util/time_guard.h`, `platform/`, `fs/`, `io/`, `column/`, `types/`, `common/`, `base/`, `gutil/`, `gen_cpp/`
 - Allowed target deps: `StoragePrimitive`, `ExecPrimitive`, `ExprCore`, `Serde`, `Util`, `RuntimeEnv`, `RuntimeCore`, `Platform`, `FSCore`, `IO`, `ChunkCore`, `ColumnCore`, `Types`, `Common`, `Base`, `Gutil`, `StarRocksGen`
 - Core tests: `compute_env_test`, `compute_env_sorting_test`
-- Remediation: Keep ComputeEnv limited to process-scoped compute resources, query-scoped scan coordination helpers without concrete scan/storage policy, query-cache primitives without concrete storage policy, WorkGroup scheduling/executor resources, shared compute-side service contracts, stable execution primitives, reusable compute-side sorting algorithms, and spill infrastructure; move concrete Exec, Storage, Service, Connector, and Agent integration upward.
+- Remediation: Keep ComputeEnv limited to process-scoped compute resources, load-path management without ExecEnv or concrete storage engine coupling, query-scoped scan coordination helpers without concrete scan/storage policy, query-cache primitives without concrete storage policy, WorkGroup scheduling/executor resources, shared compute-side service contracts, stable execution primitives, reusable compute-side sorting algorithms, and spill infrastructure; move concrete Exec, Storage, Service, Connector, and Agent integration upward.
 
 ### ExprCore (`exprcore`)
 Core expression infrastructure that depends only on RuntimeCore and lower layers.

be/CMakeLists.txt

@@ -312,6 +312,12 @@ if (CMAKE_CXX_COMPILER_ID MATCHES "Clang")
     endif()
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-unused-parameter -Wno-documentation -Wno-weak-vtables")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-unknown-warning-option")
+    # GCC 15's libstdc++ annotates __find_if (used by std::find_if/any_of/all_of/remove_if) with
+    # "#pragma GCC unroll 4". clang treats that as an explicit unroll request and emits
+    # -Wpass-failed=transform-warning when the predicate is not trivially unrollable, which -Werror
+    # turns into a build failure. These are missed-optimization diagnostics, not correctness issues,
+    # so keep the warning but do not let it fail the build.
+    set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-error=pass-failed")
     # Turn on following warning as error explicitly
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Werror=string-plus-int")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Werror=pessimizing-move")
@@ -545,15 +551,18 @@ set(STARROCKS_LINK_LIBS
     ${WL_WHOLE_ARCHIVE_PREFIX} Exprs ${WL_WHOLE_ARCHIVE_SUFFIX}
     FileSystem
     Formats
-    FormatsIO
     Storage
     Rowset
     Runtime
     Udf
     Util
     Script
-    orc
     ${WL_END_GROUP}
+    FormatAvro
+    FormatCsv
+    FormatJson
+    FormatOrc
+    FormatCore
     ExecRuntime
     ComputeEnv
     ExecPrimitive

be/module_boundary_manifest.json

@@ -264,7 +264,6 @@
       ],
       "forbidden_includes": [
         "runtime/arrow_result_writer.h",
-        "runtime/base_load_path_mgr.h",
         "runtime/broker_mgr.h",
         "runtime/buffer_control_block.h",
         "runtime/command_executor.h",
@@ -273,7 +272,6 @@
         "runtime/data_stream_recvr.h",
         "runtime/data_stream_sender.h",
         "runtime/diagnose_daemon.h",
-        "runtime/dummy_load_path_mgr.h",
         "runtime/exec_env.h",
         "runtime/export_sink.h",
         "runtime/external_scan_context_mgr.h",
@@ -287,7 +285,6 @@
         "runtime/lake_tablets_channel.h",
         "runtime/load_channel.h",
         "runtime/load_channel_mgr.h",
-        "runtime/load_path_mgr.h",
         "runtime/local_pass_through_buffer.h",
         "runtime/lookup_stream_mgr.h",
         "runtime/message_body_sink.h",
@@ -312,16 +309,67 @@
       "remediation": "Keep RuntimeCore restricted to core runtime infrastructure; move service/storage/stream-load/integration code into Runtime."
     },
     {
-      "id": "formatsio",
-      "doc_label": "FormatsIO",
-      "summary": "Format-oriented output stream primitives above RuntimeCore, FSCore, and Types.",
-      "owned_targets": ["FormatsIO"],
-      "owned_globs": ["be/src/formats/io/**"],
-      "allowed_include_prefixes": ["formats/io/", "runtime/", "fs/", "types/", "common/", "base/", "gutil/", "gen_cpp/"],
-      "allowed_target_deps": ["RuntimeCore", "FSCore", "Types", "Common", "Base", "Gutil", "StarRocksGen"],
-      "allowed_test_targets": ["formats_io_test"],
-      "allowed_test_link_deps": ["FormatsIO", "RuntimeCore", "FSCore", "Types", "Common", "Base", "Gutil", "StarRocksGen"],
-      "remediation": "Keep FormatsIO limited to reusable format output stream primitives; move connector orchestration and higher execution policy upward."
+      "id": "formatcore",
+      "doc_label": "FormatCore",
+      "summary": "Format-oriented core primitives above ComputeEnv, ExprCore, RuntimeCore, FSCore, ChunkCore, ColumnCore, and Types.",
+      "owned_targets": ["FormatCore"],
+      "owned_globs": [
+        "be/src/formats/column_evaluator.*",
+        "be/src/formats/deletion_bitmap.*",
+        "be/src/formats/disk_range.hpp",
+        "be/src/formats/file_writer.*",
+        "be/src/formats/io/**",
+        "be/src/formats/utils.h"
+      ],
+      "allowed_include_prefixes": [
+        "formats/column_evaluator.h",
+        "formats/deletion_bitmap.h",
+        "formats/disk_range.hpp",
+        "formats/file_writer.h",
+        "formats/io/",
+        "formats/utils.h",
+        "exprs/",
+        "runtime/",
+        "fs/",
+        "column/",
+        "types/",
+        "common/",
+        "base/",
+        "gutil/",
+        "gen_cpp/"
+      ],
+      "allowed_target_deps": [
+        "ComputeEnv",
+        "ExprCore",
+        "RuntimeCore",
+        "FSCore",
+        "ChunkCore",
+        "ColumnCore",
+        "Types",
+        "Common",
+        "Base",
+        "Gutil",
+        "StarRocksGen"
+      ],
+      "allowed_test_targets": ["format_test"],
+      "allowed_test_link_deps": [
+        "FormatOrc",
+        "FormatAvro",
+        "FormatJson",
+        "FormatCsv",
+        "FormatCore",
+        "ExprCore",
+        "RuntimeCore",
+        "FSCore",
+        "ChunkCore",
+        "ColumnCore",
+        "Types",
+        "Common",
+        "Base",
+        "Gutil",
+        "StarRocksGen"
+      ],
+      "remediation": "Keep FormatCore limited to reusable format primitives; move connector orchestration and higher execution policy upward."
     },
     {
       "id": "runtimeenv",
@@ -389,7 +437,7 @@
     {
       "id": "computeenv",
       "doc_label": "ComputeEnv",
-      "summary": "Shared compute-side BE environment boundary for process-scoped compute resources, query-scoped scan coordination helpers, query-cache primitives, WorkGroup scheduling/executor resources, and pipeline controls below full Exec/Storage and above RuntimeEnv.",
+      "summary": "Shared compute-side BE environment boundary for process-scoped compute resources, load-path management, query-scoped scan coordination helpers, query-cache primitives, WorkGroup scheduling/executor resources, and pipeline controls below full Exec/Storage and above RuntimeEnv.",
       "owned_targets": ["ComputeEnv"],
       "owned_roots": ["be/src/compute_env"],
       "allowed_include_prefixes": [
@@ -428,7 +476,7 @@
       "allowed_target_deps": ["StoragePrimitive", "ExecPrimitive", "ExprCore", "Serde", "Util", "RuntimeEnv", "RuntimeCore", "Platform", "FSCore", "IO", "ChunkCore", "ColumnCore", "Types", "Common", "Base", "Gutil", "StarRocksGen"],
       "allowed_test_targets": ["compute_env_test", "compute_env_sorting_test"],
       "allowed_test_link_deps": ["ComputeEnv", "ExprCore", "RuntimeCore", "FSCore", "IO", "ChunkCore", "ColumnCore", "Types", "Common", "Base", "Gutil", "StarRocksGen"],
-      "remediation": "Keep ComputeEnv limited to process-scoped compute resources, query-scoped scan coordination helpers without concrete scan/storage policy, query-cache primitives without concrete storage policy, WorkGroup scheduling/executor resources, shared compute-side service contracts, stable execution primitives, reusable compute-side sorting algorithms, and spill infrastructure; move concrete Exec, Storage, Service, Connector, and Agent integration upward."
+      "remediation": "Keep ComputeEnv limited to process-scoped compute resources, load-path management without ExecEnv or concrete storage engine coupling, query-scoped scan coordination helpers without concrete scan/storage policy, query-cache primitives without concrete storage policy, WorkGroup scheduling/executor resources, shared compute-side service contracts, stable execution primitives, reusable compute-side sorting algorithms, and spill infrastructure; move concrete Exec, Storage, Service, Connector, and Agent integration upward."
     },
     {
       "id": "exprcore",

be/src/base/string/utf8.h

@@ -14,6 +14,8 @@
 
 #pragma once
 
+#include <algorithm>
+#include <cstdint>
 #include <cstring>
 #include <vector>
 
@@ -150,4 +152,33 @@ static inline bool validate_ascii_fast(const char* src, size_t len) {
     return simdutf::validate_ascii(src, len);
 }
 
+// Returns the number of trailing bytes of [data, data+len) that form an incomplete
+// multi-byte UTF-8 sequence, i.e. a leading byte followed by fewer continuation bytes
+// than its encoded width requires. Returns 0 when the buffer already ends on a UTF-8
+// character boundary, or when the trailing bytes are not a clean truncation of a valid
+// sequence (e.g. a stray continuation byte or an invalid leading byte) -- those are
+// left for the downstream validator to reject.
+//
+// This is meant for inputs read in fixed-size chunks and handed to a UTF-8-validating
+// parser (e.g. simdjson's iterate_many, which raises UTF8_ERROR on a partial trailing
+// character): the returned bytes should be held back and prepended to the next chunk so
+// the split character can be completed.
+static inline size_t incomplete_trailing_utf8_len(const char* data, size_t len) {
+    // A UTF-8 character is at most 4 bytes, so an incomplete trailing sequence spans at
+    // most 3 bytes; scanning back at most 4 bytes always reaches its leading byte.
+    const size_t lookback = std::min<size_t>(len, 4);
+    for (size_t i = 1; i <= lookback; ++i) {
+        const auto b = static_cast<uint8_t>(data[len - i]);
+        if ((b & 0xC0) == 0x80) {
+            // continuation byte (10xxxxxx); keep scanning back for the leading byte.
+            continue;
+        }
+        // `b` is a leading byte (or ASCII) and `i` bytes of this character are present.
+        // If fewer than its encoded width are present, the trailing sequence is incomplete.
+        const size_t width = UTF8_BYTE_LENGTH_TABLE[b];
+        return i < width ? i : 0;
+    }
+    return 0;
+}
+
 } // namespace starrocks

be/src/common/config.h

@@ -1499,6 +1499,10 @@ CONF_mInt64(lake_put_txn_log_timeout_guard_ms, "-1");
 CONF_mString(lake_vacuum_retry_pattern, "*request rate*");
 CONF_mInt64(lake_vacuum_retry_max_attempts, "5");
 CONF_mInt64(lake_vacuum_retry_min_delay_ms, "100");
+// Whether vacuum tasks honor the timeout carried in the request (VacuumRequest.timeout_ms)
+// and abort themselves once it elapses. Set to false to let vacuum tasks always run to
+// completion no matter how long the FE caller waits.
+CONF_mBool(lake_vacuum_enable_task_timeout, "true");
 CONF_mInt64(lake_max_garbage_version_distance, "100");
 CONF_mBool(enable_primary_key_recover, "false");
 CONF_mBool(lake_enable_compaction_async_write, "false");

be/src/common/config_fwd_headers_manifest.json

@@ -858,6 +858,7 @@
         "lake_vacuum_retry_pattern",
         "lake_vacuum_retry_max_attempts",
         "lake_vacuum_retry_min_delay_ms",
+        "lake_vacuum_enable_task_timeout",
         "lake_max_garbage_version_distance",
         "enable_strict_delvec_crc_check",
         "lake_clear_corrupted_cache_meta",

be/src/common/config_lake_fwd.h

@@ -83,6 +83,11 @@ CONF_mInt64(lake_vacuum_retry_max_attempts, "5");
 
 CONF_mInt64(lake_vacuum_retry_min_delay_ms, "100");
 
+// Whether vacuum tasks honor the timeout carried in the request (VacuumRequest.timeout_ms)
+// and abort themselves once it elapses. Set to false to let vacuum tasks always run to
+// completion no matter how long the FE caller waits.
+CONF_mBool(lake_vacuum_enable_task_timeout, "true");
+
 CONF_mInt64(lake_max_garbage_version_distance, "100");
 
 // Enable cleanup of orphan delvec entries during compaction.

be/src/compute_env/CMakeLists.txt

@@ -79,6 +79,13 @@ set(COMPUTE_ENV_GLOBAL_DICT_FILES
     global_dict/parser.cpp
 )
 
+set(COMPUTE_ENV_LOAD_PATH_FILES
+    load_path/load_path_state_helper.cpp
+    load_path/load_path_mgr.cpp
+    load_path/dummy_load_path_mgr.cpp
+    load_path/rejected_record_writer.cpp
+)
+
 ADD_BE_LIB(ComputeEnv
     compute_env.cpp
     data_stream/data_stream_mgr.cpp
@@ -98,6 +105,7 @@ ADD_BE_LIB(ComputeEnv
     ${COMPUTE_ENV_QUERY_CACHE_FILES}
     ${COMPUTE_ENV_QUERY_FILES}
     ${COMPUTE_ENV_GLOBAL_DICT_FILES}
+    ${COMPUTE_ENV_LOAD_PATH_FILES}
 )
 
 target_link_libraries(ComputeEnv PUBLIC

be/src/compute_env/compute_env.cpp

@@ -22,6 +22,8 @@
 #include "common/logging.h"
 #include "common/system/cpu_info.h"
 #include "compute_env/data_stream/data_stream_mgr.h"
+#include "compute_env/load_path/dummy_load_path_mgr.h"
+#include "compute_env/load_path/load_path_mgr.h"
 #include "compute_env/pipeline/driver_limiter.h"
 #include "compute_env/pipeline/pipeline_timer.h"
 #include "compute_env/profile_report_worker.h"
@@ -93,6 +95,21 @@ Status ComputeEnv::init_workgroup(const ComputeEnvWorkGroupOptions& options) {
     return Status::OK();
 }
 
+Status ComputeEnv::init_load_path(std::vector<std::string> store_paths, bool use_dummy_load_path_mgr) {
+    if (_load_path_mgr != nullptr) {
+        return Status::InternalError("LoadPathMgr has been initialized");
+    }
+    std::unique_ptr<BaseLoadPathMgr> load_path_mgr;
+    if (use_dummy_load_path_mgr) {
+        load_path_mgr = std::make_unique<DummyLoadPathMgr>();
+    } else {
+        load_path_mgr = std::make_unique<LoadPathMgr>(std::move(store_paths));
+    }
+    RETURN_IF_ERROR(load_path_mgr->init());
+    _load_path_mgr = std::move(load_path_mgr);
+    return Status::OK();
+}
+
 Status ComputeEnv::init_spill(const std::vector<std::string>& store_paths, MetricRegistry* metrics) {
     auto spill_dir_mgr = std::make_shared<spill::DirManager>();
     RETURN_IF_ERROR(spill_dir_mgr->init(config::spill_local_storage_dir, store_paths));
@@ -168,8 +185,13 @@ void ComputeEnv::destroy_profile_report_worker() {
     _profile_report_worker.reset();
 }
 
+void ComputeEnv::destroy_load_path() {
+    _load_path_mgr.reset();
+}
+
 void ComputeEnv::destroy() {
     destroy_profile_report_worker();
+    destroy_load_path();
     _global_spill_manager.reset();
     _spill_dir_mgr.reset();
     if (_workgroup_manager != nullptr) {

be/src/compute_env/compute_env.h

@@ -17,13 +17,15 @@
 #include <cstddef>
 #include <cstdint>
 #include <memory>
+#include <string>
 #include <vector>
 
 #include "common/status.h"
 #include "compute_env/workgroup/work_group_manager.h"
 
 namespace starrocks {
 
+class BaseLoadPathMgr;
 class DataStreamMgr;
 class MetricRegistry;
 class ProfileReportWorker;
@@ -68,6 +70,7 @@ class ComputeEnv {
 
     Status init(const ComputeEnvOptions& options);
     Status init_workgroup(const ComputeEnvWorkGroupOptions& options);
+    Status init_load_path(std::vector<std::string> store_paths, bool use_dummy_load_path_mgr);
     Status init_spill(const std::vector<std::string>& store_paths, MetricRegistry* metrics);
     Status init_query_cache(size_t capacity);
     Status init_profile_report_worker(ProfileReportWorkerOptions options);
@@ -77,6 +80,7 @@ class ComputeEnv {
     Status start_result_mgr();
     void stop_result_mgr();
     void destroy_profile_report_worker();
+    void destroy_load_path();
     void destroy();
 
     pipeline::DriverLimiter* driver_limiter() const { return _driver_limiter.get(); }
@@ -89,6 +93,7 @@ class ComputeEnv {
     spill::GlobalSpillManager* global_spill_manager() const { return _global_spill_manager.get(); }
     query_cache::CacheManagerRawPtr cache_mgr() const { return _cache_mgr.get(); }
     ProfileReportWorker* profile_report_worker() const { return _profile_report_worker.get(); }
+    BaseLoadPathMgr* load_path_mgr() const { return _load_path_mgr.get(); }
 
 private:
     std::unique_ptr<pipeline::DriverLimiter> _driver_limiter;
@@ -101,6 +106,7 @@ class ComputeEnv {
     std::shared_ptr<spill::GlobalSpillManager> _global_spill_manager;
     std::unique_ptr<query_cache::CacheManager> _cache_mgr;
     std::unique_ptr<ProfileReportWorker> _profile_report_worker;
+    std::unique_ptr<BaseLoadPathMgr> _load_path_mgr;
 };
 
 } // namespace starrocks