HappenLee commented on code in PR #67762:
URL: https://github.com/apache/doris/pull/67762#discussion_r4313174588
##########
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:
Fixed in ad135bf83bfcdbe6995b1cdb2229e51da3119b8f.
The dispatch now moves the first error through a noexcept path and retains a
preallocated cancellation status if reporting an error itself throws. Rejected
submissions retire through RAII, and post-admission exceptions retire the
submitter through the dispatch instead of unwinding into premature RPC
completion. The parent error path releases only its shared guard ownership;
accepted reads retain the RPC. FIFO dequeue now moves the published callable
instead of copying it.
Validation: 59 related ASAN BE unit tests passed, including an injected
exception at error publication with accepted workers, the same fault with an
active worker in a real saturated FIFO, and a callable whose copy constructor
throws after publication. Completion occurs once and only after accepted reads
exit.
##########
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:
This finding is not fixed by ad135bf83bfcdbe6995b1cdb2229e51da3119b8f; that
revision addresses P1-1 and P1-3 only.
The BE-wide rowid pool still does not use the request Workload Group's
CgroupCpuCtl or per-group admission limits, including the default serial path.
The new external-rejection test proves that a rejected child no longer pins a
shared worker; it does not establish Workload Group CPU/admission isolation.
This concern still needs a separate execution/admission change and a two-group
isolation test.
##########
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:
Fixed in ad135bf83bfcdbe6995b1cdb2229e51da3119b8f.
TimeSharingTaskExecutor now propagates admission failures and retires
rejected splits from executor/handle running state, closing their futures
outside the executor mutex. The one-callback scanner entry also removes the
rejected task handle. Future-result bookkeeping is allocated before admission,
and a queued split is withdrawn before reporting failure to create its first
worker thread.
Validation: 59 related ASAN BE unit tests passed. The real-scheduler
saturation test accepts one child and rejects the next; the parent waits for
the accepted child, returns an error, and a subsequent request on the shared
FIFO progresses. Additional tests cover stopped schedulers, closed task
handles, and first-thread creation failure without later execution of the
rejected split.
--
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]
