This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 8134f160240 branch-4.2: [improvement](be) Use thread pool scan
scheduler by default #68610 (#69508)
8134f160240 is described below
commit 8134f1602407c04e3a43bb89a7b91ab09762ab7f
Author: Jerry Hu <[email protected]>
AuthorDate: Sat Oct 24 20:53:15 2026 +0800
branch-4.2: [improvement](be) Use thread pool scan scheduler by default
#68610 (#69508)
## Summary
- Use the existing thread-pool scheduler by default for internal and
external scans. The task executor remains available through the
corresponding explicit configuration options.
- Convert Doris exceptions raised during row-ID V2 fetch RPC processing
on a thread-pool worker into an error status in the RPC response.
## Backport scope
Cherry-picked from #68610 for branch-4.2. This branch does not have the
master-side shared scan LIMIT counter or the associated scanner/context
APIs, so its existing scanner behavior is retained. The backport
includes the scheduler defaults, the row-ID RPC exception boundary, and
focused tests without importing unrelated master-side scan changes.
## Tests
- ASAN BE and FE build: `BUILD_TYPE=ASAN ./build.sh --be --fe -j 32` —
passed.
- Focused BE unit tests covering row-ID fetch, external scan task
submission, workload-group schedulers, scanner context, and scanner
projection — 82 passed.
- `query_p0/limit` regression on an isolated local FE/BE cluster — 18
suites passed.
- `clang-format.sh`, `check-format.sh`, and `git diff --check` — passed.
## Release note
Internal and external scans now use the thread-pool scheduler by
default. Set `enable_task_executor_in_internal_table` or
`enable_task_executor_in_external_table` to `true` to opt back into the
task executor for the respective scan type.
---
be/src/common/config.cpp | 4 +-
be/src/exec/rowid_fetcher.cpp | 7 +++
be/src/exec/rowid_fetcher.h | 4 ++
be/src/runtime/exec_env.h | 1 +
be/src/service/internal_service.cpp | 4 +-
be/test/exec/rowid_fetcher_test.cpp | 73 ++++++++++++++++++++++
.../workload_group/workload_group_manager_test.cpp | 57 +++++++++++++++++
7 files changed, 145 insertions(+), 5 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 62476f51bc0..8c1c0f62a01 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -330,9 +330,9 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
// Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "true");
+DEFINE_Bool(enable_task_executor_in_internal_table, "false");
// Enable task executor in external table scan.
-DEFINE_Bool(enable_task_executor_in_external_table, "true");
+DEFINE_Bool(enable_task_executor_in_external_table, "false");
// number of scanner thread pool size for olap table
// and the min thread num of remote scanner thread pool
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index 5fc288dd301..32de6d7dfd5 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -559,6 +559,13 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequest& request,
return Status::OK();
}
+void RowIdStorageReader::read_by_rowids_for_rpc(const PMultiGetRequestV2&
request,
+ PMultiGetResponseV2* response)
{
+ Status st = Status::OK();
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(st = read_by_rowids(request, response),
st);
+ st.to_protobuf(response->mutable_status());
+}
+
Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response) {
if (request.request_block_descs_size()) {
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index ca1f8b50690..f4c400f0b07 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -127,6 +127,10 @@ public:
static Status read_by_rowids(const PMultiGetRequest& request,
PMultiGetResponse* response);
static Status read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response);
+ // ThreadPool scan workers have no exception boundary; keep RPC failures
in the response.
+ static void read_by_rowids_for_rpc(const PMultiGetRequestV2& request,
+ PMultiGetResponseV2* response);
+
static bool should_use_file_scanner_v2(const TQueryOptions& query_options,
const TFileScanRangeParams&
scan_params,
const TFileRangeDesc& range);
diff --git a/be/src/runtime/exec_env.h b/be/src/runtime/exec_env.h
index 69160d88f42..97424165056 100644
--- a/be/src/runtime/exec_env.h
+++ b/be/src/runtime/exec_env.h
@@ -339,6 +339,7 @@ public:
void set_cache_manager(CacheManager* cm) { this->_cache_manager = cm; }
void set_process_profile(ProcessProfile* pp) { this->_process_profile =
pp; }
void set_tablet_schema_cache(TabletSchemaCache* c) {
this->_tablet_schema_cache = c; }
+ void set_id_manager(IdManager* id_manager) { this->_id_manager =
id_manager; }
void set_delete_bitmap_agg_cache(DeleteBitmapAggCache* c) {
_delete_bitmap_agg_cache = c; }
void set_tablet_column_object_pool(TabletColumnObjectPool* c) {
this->_tablet_column_object_pool = c;
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index 3f8719acce7..89d7e0de834 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -2335,9 +2335,7 @@ void
PInternalService::multiget_data_v2(google::protobuf::RpcController* control
MonotonicStopWatch watch;
watch.start();
brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- Status st =
RowIdStorageReader::read_by_rowids(*request, response);
- st.to_protobuf(response->mutable_status());
+ RowIdStorageReader::read_by_rowids_for_rpc(*request,
response);
LOG(INFO) << "multiget_data finished, cost(us):"
<< watch.elapsed_time() / 1000;
return true;
diff --git a/be/test/exec/rowid_fetcher_test.cpp
b/be/test/exec/rowid_fetcher_test.cpp
index 3e8fd3e9517..e4246c4974f 100644
--- a/be/test/exec/rowid_fetcher_test.cpp
+++ b/be/test/exec/rowid_fetcher_test.cpp
@@ -19,17 +19,27 @@
#include <gtest/gtest.h>
+#include <chrono>
+#include <memory>
#include <string>
#include <vector>
#include "common/consts.h"
+#include "common/exception.h"
#include "exec/operator/file_scan_operator.h"
#include "exec/scan/file_scanner_v2.h"
+#include "exec/scan/scanner_scheduler.h"
#include "format_v2/column_mapper.h"
#include "format_v2/table/hive_reader.h"
+#include "gen_cpp/internal_service.pb.h"
#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
+#include "storage/id_manager.h"
+#include "storage/olap_common.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
namespace doris {
@@ -423,4 +433,67 @@ TEST_F(SubmitExternalScanTasksTest,
ThrownExceptionReachesTheCaller) {
EXPECT_NE(result.to_string().find("scanner threw"), std::string::npos);
}
+// multiget_data_v2 runs read_by_rowids() on the remote scan scheduler, which
by default is a
+// ThreadPool whose worker has no exception boundary. A request whose file
mapping exists but
+// whose slot carries a PTypeNode this BE cannot decode makes SlotDescriptor's
constructor throw;
+// the RPC helper must turn that into an error status on the response instead
of letting the
+// exception leave the worker and terminate the process.
+TEST_F(RowIdStorageReaderTest,
RpcHelperReportsThrownExceptionOnThreadPoolPath) {
+ IdManager id_manager;
+ ExecEnv* env = ExecEnv::GetInstance();
+ IdManager* saved_id_manager = env->get_id_manager();
+ env->set_id_manager(&id_manager);
+ Defer restore_id_manager {[&] { env->set_id_manager(saved_id_manager); }};
+
+ PMultiGetRequestV2 request;
+ request.mutable_query_id()->set_hi(68610);
+ request.mutable_query_id()->set_lo(1);
+ request.set_wg_id(1);
+ auto id_file_map = id_manager.add_id_file_map(request.query_id(), 1024);
+ const uint32_t file_id = id_file_map->get_file_mapping_id(
+ std::make_shared<FileMapping>(/*tablet_id=*/1, RowsetId(),
/*segment_id=*/0));
+
+ PRequestBlockDesc* block_desc = request.add_request_block_descs();
+ block_desc->add_file_id(file_id);
+ block_desc->add_row_id(0);
+ PSlotDescriptor* slot = block_desc->add_slots();
+ slot->set_id(0);
+ slot->set_parent(0);
+ slot->set_column_pos(0);
+ slot->set_byte_offset(0);
+ slot->set_null_indicator_byte(0);
+ slot->set_null_indicator_bit(-1);
+ slot->set_col_name("c");
+ slot->set_slot_idx(0);
+ // A type node kind unknown to this BE, as a newer FE could send.
+ slot->mutable_slot_type()->add_types()->set_type(99);
+
+ // The worker captures the response and the latch by reference. Declare
them before the
+ // scheduler so every exit, including a failed wait below, joins the
worker in stop_scheduler
+ // before either captured object is destroyed.
+ PMultiGetResponseV2 response;
+ CountDownLatch finished(1);
+ std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl =
std::make_shared<CgroupV2CpuCtl>(1);
+ ThreadPoolSimplifiedScanScheduler scheduler("rowid_rpc_test",
cgroup_cpu_ctl);
+ ASSERT_TRUE(scheduler.start(1, 1, 4, 1).ok());
+ Defer stop_scheduler {[&] { scheduler.stop(); }};
+
+ ASSERT_TRUE(scheduler
+ .submit_scan_task(SimplifiedScanTask(
+ [&]() {
+
RowIdStorageReader::read_by_rowids_for_rpc(
+ request,
&response);
+ finished.count_down();
+ return true;
+ },
+ nullptr, nullptr),
+ "rowid_rpc_test_task")
+ .ok());
+ ASSERT_TRUE(finished.wait_for(std::chrono::seconds(30)));
+
+ EXPECT_NE(response.status().status_code(), 0);
+ ASSERT_EQ(response.status().error_msgs_size(), 1);
+ EXPECT_NE(response.status().error_msgs(0).find("invalid TTypeNodeType"),
std::string::npos);
+}
+
} // namespace doris
diff --git a/be/test/runtime/workload_group/workload_group_manager_test.cpp
b/be/test/runtime/workload_group/workload_group_manager_test.cpp
index 8a10c619bfd..2a61c1319c5 100644
--- a/be/test/runtime/workload_group/workload_group_manager_test.cpp
+++ b/be/test/runtime/workload_group/workload_group_manager_test.cpp
@@ -36,6 +36,7 @@
#include "cpp/sync_point.h"
#include "exec/pipeline/dependency.h"
#include "exec/pipeline/pipeline_tracing.h"
+#include "exec/scan/scanner_scheduler.h"
#include "exec/spill/spill_file_manager.h"
#include "runtime/exec_env.h"
#include "runtime/query_context.h"
@@ -1360,4 +1361,60 @@ TEST_F(WorkloadGroupManagerTest,
AdaptiveFlushRegistrationSurvivesIdChangeAndReu
controller->adjust_once();
}
+TEST_F(WorkloadGroupManagerTest, DefaultScanSchedulersUseThreadPool) {
+ // Check the registered defaults, which do not depend on the be.conf
loaded by the UT runner.
+ const auto& fields = *config::Register::_s_field_map;
+ EXPECT_STREQ("false",
fields.at("enable_task_executor_in_internal_table").defval);
+ EXPECT_STREQ("false",
fields.at("enable_task_executor_in_external_table").defval);
+
+ const auto saved_config = std::make_tuple(
+ config::enable_adaptive_flush_threads,
config::enable_task_executor_in_internal_table,
+ config::enable_task_executor_in_external_table,
config::pipeline_executor_size,
+ config::blocking_pipeline_executor_size,
config::doris_scanner_thread_pool_thread_num,
+ config::doris_max_remote_scanner_thread_pool_thread_num,
+ config::doris_scanner_min_thread_pool_thread_num,
config::min_active_scan_threads,
+ config::min_active_file_scan_threads,
config::flush_thread_num_per_store);
+ Defer restore {[&] {
+ std::tie(config::enable_adaptive_flush_threads,
+ config::enable_task_executor_in_internal_table,
+ config::enable_task_executor_in_external_table,
config::pipeline_executor_size,
+ config::blocking_pipeline_executor_size,
+ config::doris_scanner_thread_pool_thread_num,
+ config::doris_max_remote_scanner_thread_pool_thread_num,
+ config::doris_scanner_min_thread_pool_thread_num,
config::min_active_scan_threads,
+ config::min_active_file_scan_threads,
config::flush_thread_num_per_store) =
+ saved_config;
+ }};
+ config::enable_adaptive_flush_threads = false;
+ config::enable_task_executor_in_internal_table = false;
+ config::enable_task_executor_in_external_table = false;
+ config::pipeline_executor_size = 1;
+ config::blocking_pipeline_executor_size = 1;
+ config::doris_scanner_thread_pool_thread_num = 1;
+ config::doris_max_remote_scanner_thread_pool_thread_num = 1;
+ config::doris_scanner_min_thread_pool_thread_num = 1;
+ config::min_active_scan_threads = 1;
+ config::min_active_file_scan_threads = 1;
+ config::flush_thread_num_per_store = 1;
+
+ ASSERT_TRUE(_wg_manager->create_internal_wg().ok());
+ Defer stop_schedulers {[&] {
+ for (auto& entry : _wg_manager->_workload_groups) {
+ entry.second->try_stop_schedulers();
+ entry.second->destroy_schedulers();
+ }
+ }};
+ ASSERT_EQ(_wg_manager->_workload_groups.size(), 1);
+ auto wg = _wg_manager->_workload_groups.begin()->second;
+
+ TaskScheduler* exec_sched = nullptr;
+ ScannerScheduler* scan_sched = nullptr;
+ ScannerScheduler* remote_scan_sched = nullptr;
+ wg->get_query_scheduler(&exec_sched, &scan_sched, &remote_scan_sched);
+ ASSERT_NE(scan_sched, nullptr);
+ ASSERT_NE(remote_scan_sched, nullptr);
+ EXPECT_NE(dynamic_cast<ThreadPoolSimplifiedScanScheduler*>(scan_sched),
nullptr);
+
EXPECT_NE(dynamic_cast<ThreadPoolSimplifiedScanScheduler*>(remote_scan_sched),
nullptr);
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
