github-actions[bot] commented on code in PR #67762:
URL: https://github.com/apache/doris/pull/67762#discussion_r4025047077
##########
be/src/service/internal_service.cpp:
##########
@@ -2006,33 +2012,54 @@ void
PInternalService::multiget_data_v2(google::protobuf::RpcController* control
return;
}
- doris::TaskScheduler* exec_sched = nullptr;
- ScannerScheduler* scan_sched = nullptr;
- ScannerScheduler* remote_scan_sched = nullptr;
- wg->get_query_scheduler(&exec_sched, &scan_sched, &remote_scan_sched);
- DCHECK(remote_scan_sched);
-
- st = remote_scan_sched->submit_scan_task(
- SimplifiedScanTask(
- [request, response, done]() {
-
SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->rowid_storage_reader_tracker());
- signal::set_signal_task_id(request->query_id());
- // multi get data by rowid
- 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());
- LOG(INFO) << "multiget_data finished, cost(us):"
- << watch.elapsed_time() / 1000;
- return true;
- },
- nullptr, nullptr),
- fmt::format("{}-multiget_data_v2", print_id(request->query_id())));
+ // Retain the RPC until every asynchronous internal read finishes.
+ auto closure_guard = std::make_shared<brpc::ClosureGuard>(done);
+ // A queued parent task may also be discarded during pool shutdown.
+ Status::Cancelled("Row id fetch task discarded before execution")
+ .to_protobuf(response->mutable_status());
+ auto task = [this, request, response, closure_guard]() {
+
SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->rowid_storage_reader_tracker());
+ signal::set_signal_task_id(request->query_id());
+ MonotonicStopWatch watch;
+ watch.start();
+ Status status;
+ try {
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(
+ RowIdStorageReader::read_by_rowids(
+ *request, response, &_rowid_fetch_pool,
+ [response, closure_guard, watch](Status result) {
+ result.to_protobuf(response->mutable_status());
+ LOG(INFO) << "multiget_data finished,
cost(us):"
+ << watch.elapsed_time() / 1000;
+ // Complete while the rowid tracker is
attached.
+ closure_guard->reset(nullptr);
+ }),
+ status);
+ } catch (const std::exception& e) {
+ status = Status::InternalError("Row id fetch failed because {}",
e.what());
+ }
+ if (!status.ok()) {
+ status.to_protobuf(response->mutable_status());
+ closure_guard->reset(nullptr);
+ }
+ };
+ // Use the dedicated pool for both preparation and internal reads.
Preparation
+ // returns after submitting its workers, so even a one-thread pool can
progress.
+ try {
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(
+ {
+ if (!_rowid_fetch_pool.try_offer(std::move(task))) {
Review Comment:
[P1] Preserve Workload Group isolation for rowid fetches. This pool is
shared by every group and its pthreads are not created with the request group's
`CgroupCpuCtl`, whereas the removed `remote_scan_sched` path used that group's
cgroup and configured thread limits. The change routes even
`parallel_batch_rows == 0` here, where `prepare_internal_block()` performs the
serial storage read, so a CPU-limited group can exceed its hard limit and a hot
group can occupy/fill the global FIFO until unrelated groups receive
`SERVICE_UNAVAILABLE`. Please keep execution/admission workload-aware
(including the default serial path) and add a two-group isolation test.
##########
be/src/service/internal_service.cpp:
##########
@@ -241,7 +241,9 @@ PInternalService::PInternalService(ExecEnv* exec_env)
config::brpc_arrow_flight_work_pool_max_queue_size != -1
?
config::brpc_arrow_flight_work_pool_max_queue_size
: std::max(20480,
CpuInfo::num_cores() * 640),
- "brpc_arrow_flight") {
+ "brpc_arrow_flight"),
+ _rowid_fetch_pool(CpuInfo::num_cores(), std::max(10240,
CpuInfo::num_cores() * 320),
Review Comment:
[P2] Expose queue-depth and active-worker metrics for this pool. Every
neighboring internal-service pool registers and deregisters those hooks, but
this global core-sized pool can block on storage/external work while its
10,240-plus queue grows with no server-side signal. `get_info()` appears only
after admission has already failed and is returned to the client, so operators
cannot alert on saturation or distinguish stuck workers from ordinary rowid
latency. Please add the matching rowid-pool gauges and deregistration.
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -312,104 +509,153 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequestV2& request,
set_topn_lazy_materialization_file_cache_stats(
stats.file_cache_stats,
response->mutable_topn_lazy_materialization_file_cache_stats());
+
+ return Status::OK();
}
+};
- return Status::OK();
-}
+namespace {
-Status RowIdStorageReader::read_batch_doris_format_row(
- const PRequestBlockDesc& request_block_desc,
std::shared_ptr<IdFileMap> id_file_map,
- std::vector<SlotDescriptor>& slots, const TUniqueId& query_id, Block&
result_block,
- OlapReaderStatistics& stats, int64_t* acquire_tablet_ms, int64_t*
acquire_rowsets_ms,
- int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms,
- io::FileCacheMissPolicy file_cache_miss_policy) {
- if (result_block.is_empty_column()) [[likely]] {
- result_block = Block(slots, request_block_desc.row_id_size());
- }
- TabletSchema full_read_schema;
- for (const ColumnPB& column_pb : request_block_desc.column_descs()) {
- full_read_schema.append_column(TabletColumn(column_pb));
+// Both storage reads and pool submission can throw. Convert exceptions on the
+// same pthread where they originated, before publishing completion.
+template <typename Func>
+Status rowid_read_status(Func&& func) {
+ Status status;
+ try {
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(status = func(), status);
+ } catch (const std::exception& e) {
+ status = Status::InternalError("Row id fetch failed because {}",
e.what());
}
+ return status;
+}
- std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>
iterator_map;
- std::unordered_map<SegKey, SegItem, HashOfSegKey> seg_map;
- std::string row_store_buffer;
- RowStoreReadStruct row_store_read_struct(row_store_buffer);
- if (request_block_desc.fetch_row_store()) {
- for (int i = 0; i < request_block_desc.slots_size(); ++i) {
-
row_store_read_struct.serdes.emplace_back(slots[i].get_data_type_ptr()->get_serde());
- row_store_read_struct.col_uid_to_idx[slots[i].col_unique_id()] = i;
-
row_store_read_struct.default_values.emplace_back(slots[i].col_default_value());
+struct RowIdReadDispatch {
+ size_t task_count = 0;
+ std::function<Status(size_t)> run_task;
+ std::function<void(Status)> on_complete;
+ std::atomic<size_t> next_task = 0;
+ // The submitting thread owns one reference until all submissions finish.
+ // Even inline/fast workers cannot finish the RPC during submission.
+ std::atomic<size_t> remaining = 1;
+ AtomicStatus status;
+
+ ~RowIdReadDispatch() {
+ // Pool shutdown may destroy queued closures without executing them.
+ // The last closure can disappear only after all running reads have
exited.
+ if (on_complete) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ std::optional<AttachTask> task_context;
+ if (!thread_context()->is_attach_task()) {
+
task_context.emplace(ExecEnv::GetInstance()->rowid_storage_reader_tracker());
+ }
+ status.update(Status::Cancelled("Row id read tasks discarded by
pool shutdown"));
+ complete();
}
}
- // Phase 1: Group all row_ids by their (tablet_id, rowset_id, segment_id)
key.
- // Unlike the old code which only batched adjacent rows with the same
file_id,
- // this merges non-contiguous same-segment requests into a single batch,
- // maximizing the number of rows read per seek_and_read_by_rowid call.
- std::vector<DorisFormatReadBatch> scan_batches;
- std::unordered_map<SegKey, size_t, HashOfSegKey> batch_idx_by_seg;
- // (batch_idx, position_in_batch) for each row in the original request.
- std::vector<std::pair<size_t, size_t>>
row_id_block_idx(request_block_desc.row_id_size());
- for (int j = 0; j < request_block_desc.row_id_size(); ++j) {
- auto file_id = request_block_desc.file_id(j);
- auto file_mapping = id_file_map->get_file_mapping(file_id);
- if (!file_mapping) {
- return Status::InternalError(
- "Backend:{} file_mapping not found, query_id: {}, file_id:
{}",
- BackendOptions::get_localhost(), print_id(query_id),
file_id);
- }
+ void complete() {
+ // Release request buffers while the rowid tracker is attached, even if
+ // the pool retains finished task closures for longer.
+ auto completion = std::move(on_complete);
+ completion(status.status());
+ run_task = {};
+ }
- // Derive segment key and group by it — rows from the same segment are
batched together
- // even if they are interleaved with rows from other segments in the
request.
- auto [tablet_id, rowset_id, segment_id] =
file_mapping->get_doris_format_info();
- SegKey seg_key {.tablet_id = tablet_id, .rowset_id = rowset_id,
.segment_id = segment_id};
- auto [it, inserted] = batch_idx_by_seg.emplace(seg_key,
scan_batches.size());
- if (inserted) {
- // First time seeing this segment, create a new batch for it.
- scan_batches.emplace_back();
- scan_batches.back().file_mapping = file_mapping;
+ void finish_worker() {
+ if (remaining.fetch_sub(1) == 1) {
+ complete();
}
- // Record (row_id, original_request_index) for later sorting and
scattering.
-
scan_batches[it->second].row_ids_with_positions.emplace_back(request_block_desc.row_id(j),
- j);
}
+};
+
+} // namespace
- // Phase 2: For each segment, sort row_ids ascending (required by
ColumnIterator),
- // deduplicate, then read all rows in a single batch call.
- std::vector<Block> scan_blocks(scan_batches.size());
- for (size_t batch_idx = 0; batch_idx < scan_batches.size(); ++batch_idx) {
- auto& scan_batch = scan_batches[batch_idx];
- auto& row_ids_with_positions = scan_batch.row_ids_with_positions;
- std::sort(row_ids_with_positions.begin(), row_ids_with_positions.end(),
- [](const auto& lhs, const auto& rhs) { return lhs.first <
rhs.first; });
-
- // Column iterators read rowids monotonically. Deduplicate consecutive
identical row_ids
- // (different file_ids may map to the same row), then scatter rows
back to their original
- // request positions.
- std::vector<uint32_t> row_ids;
- row_ids.reserve(row_ids_with_positions.size());
-
- // Also builds the scatter map: row_id_block_idx[original_request_idx]
->
- // (batch_idx, deduplicated_position_in_batch).
- for (const auto& [row_id, result_idx] : row_ids_with_positions) {
- if (row_ids.empty() || row_ids.back() != row_id) {
- row_ids.emplace_back(row_id);
+void RowIdStorageReader::submit_internal_read_tasks(FifoThreadPool* pool,
size_t task_count,
+ int concurrency,
+
std::function<Status(size_t)> run_task,
+
std::function<void(Status)> on_complete) {
+ std::shared_ptr<RowIdReadDispatch> dispatch;
+ auto setup_status = rowid_read_status([&]() {
+ dispatch = std::make_shared<RowIdReadDispatch>();
+ dispatch->task_count = task_count;
+ dispatch->run_task = std::move(run_task);
+ dispatch->on_complete = std::move(on_complete);
+ return Status::OK();
+ });
+ if (!setup_status.ok()) {
+ on_complete(setup_status);
+ return;
+ }
+ // Launch only a bounded number of workers, each pulling the next segment
range.
+ // Never block a pool thread waiting for work submitted to its own pool.
+ DCHECK_GT(concurrency, 0);
+ const size_t workers = std::min(task_count,
static_cast<size_t>(concurrency));
+ for (size_t i = 0; i < workers; ++i) {
+ dispatch->remaining.fetch_add(1);
+ auto status = rowid_read_status([&]() {
+ // FifoThreadPool has no fallible bookkeeping after publishing a
task.
+ // A rejection or exception therefore means this worker will never
run.
+ if (!pool->try_offer([dispatch]() {
+ std::optional<AttachTask> task_context;
+ dispatch->status.update(rowid_read_status([&]() {
+ task_context.emplace(
+
ExecEnv::GetInstance()->rowid_storage_reader_tracker());
+ while (dispatch->status.ok()) {
+ const auto idx = dispatch->next_task.fetch_add(1);
+ if (idx >= dispatch->task_count) {
+ break;
+ }
+ RETURN_IF_ERROR(dispatch->run_task(idx));
+ }
+ return Status::OK();
+ }));
+ dispatch->finish_worker();
+ })) {
+ return Status::Error<ErrorCode::SERVICE_UNAVAILABLE>(
+ "Row id fetch queue full or pool stopped: {}",
pool->get_info());
}
- row_id_block_idx[result_idx] = std::make_pair(batch_idx,
row_ids.size() - 1);
+ return Status::OK();
+ });
+ if (!status.ok()) {
+ dispatch->status.update(status);
Review Comment:
[P1] Keep completion owned by the dispatch if rejection reporting throws.
After an earlier `try_offer` succeeds, this `AtomicStatus::update(status)`
copies the status error message and can allocate. If that copy throws, both
this `finish_worker()` and the submitter's final retirement are skipped; the
exception reaches the parent RPC catch, which resets `ClosureGuard` while the
accepted worker still holds `ReadRequestState` references into the BRPC
request. When that worker runs it can dereference freed request data. This is
distinct from the old TaskExecutor window, and the new tests only inject inside
`try_offer`. Make post-admission status publication/count retirement
non-throwing or RAII-owned, and fault-inject this exact point. Also note that
`BlockingQueue::controlled_blocking_get()` currently copies an
already-published `std::function`; moving it or adding a no-throw completion
guard is needed to remove the remaining post-publication allocation.
##########
be/src/service/internal_service.cpp:
##########
@@ -2006,33 +2012,54 @@ void
PInternalService::multiget_data_v2(google::protobuf::RpcController* control
return;
}
- doris::TaskScheduler* exec_sched = nullptr;
- ScannerScheduler* scan_sched = nullptr;
- ScannerScheduler* remote_scan_sched = nullptr;
- wg->get_query_scheduler(&exec_sched, &scan_sched, &remote_scan_sched);
- DCHECK(remote_scan_sched);
-
- st = remote_scan_sched->submit_scan_task(
- SimplifiedScanTask(
- [request, response, done]() {
-
SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->rowid_storage_reader_tracker());
- signal::set_signal_task_id(request->query_id());
- // multi get data by rowid
- 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());
- LOG(INFO) << "multiget_data finished, cost(us):"
- << watch.elapsed_time() / 1000;
- return true;
- },
- nullptr, nullptr),
- fmt::format("{}-multiget_data_v2", print_id(request->query_id())));
+ // Retain the RPC until every asynchronous internal read finishes.
+ auto closure_guard = std::make_shared<brpc::ClosureGuard>(done);
+ // A queued parent task may also be discarded during pool shutdown.
+ Status::Cancelled("Row id fetch task discarded before execution")
+ .to_protobuf(response->mutable_status());
+ auto task = [this, request, response, closure_guard]() {
+
SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->rowid_storage_reader_tracker());
+ signal::set_signal_task_id(request->query_id());
+ MonotonicStopWatch watch;
+ watch.start();
+ Status status;
+ try {
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(
+ RowIdStorageReader::read_by_rowids(
+ *request, response, &_rowid_fetch_pool,
Review Comment:
[P1] Do not let external fetches pin this BE-wide pool on the scheduler
admission bug. `prepare()` still calls `read_batch_external_row()`
synchronously; its children use the default
`TaskExecutorSimplifiedScanScheduler` and `submit_external_scan_tasks()` waits
for every submission it was told succeeded. As documented in the existing
admission thread, that scheduler can return OK after `_do_submit()` rejected
the split, so this wait never finishes. Previously the stuck parent consumed
only its Workload Group's remote-scan worker; now `CpuInfo::num_cores()` such
external requests occupy every global rowid worker and block all groups. Keep
the external parent isolated to its WG or make that child admission
observable/cancellable, and add a rejection test that proves another group
still progresses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
