github-actions[bot] commented on code in PR #67762:
URL: https://github.com/apache/doris/pull/67762#discussion_r4024156630
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -312,104 +508,152 @@ 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 {
+
+// Both storage reads and scheduler 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;
}
-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());
+struct RowIdScanDispatch {
+ 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;
+
+ ~RowIdScanDispatch() {
+ // Scheduler 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 scan tasks discarded by
scheduler shutdown"));
+ complete();
+ }
}
- TabletSchema full_read_schema;
- for (const ColumnPB& column_pb : request_block_desc.column_descs()) {
- full_read_schema.append_column(TabletColumn(column_pb));
+
+ void complete() {
+ // Release request buffers while the rowid tracker is attached, even if
+ // the scheduler retains finished task closures for longer.
+ auto completion = std::move(on_complete);
+ completion(status.status());
+ run_task = {};
}
- 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());
+ void finish_worker() {
+ if (remaining.fetch_sub(1) == 1) {
+ 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);
- }
+} // namespace
- // 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 RowIdStorageReader::submit_internal_scan_tasks(ScannerScheduler*
scheduler, size_t task_count,
+ int concurrency,
+
std::function<Status(size_t)> run_task,
+
std::function<void(Status)> on_complete) {
+ std::shared_ptr<RowIdScanDispatch> dispatch;
+ auto setup_status = rowid_read_status([&]() {
+ dispatch = std::make_shared<RowIdScanDispatch>();
+ 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 scanner thread waiting for work submitted to its own
scheduler.
+ 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([&]() {
+ return scheduler->submit_scan_task(
Review Comment:
[P1] Propagate the task-executor admission failure. The default remote
scheduler is `TaskExecutorSimplifiedScanScheduler`: `enqueue_splits()` can move
this worker into the handle's running set and call `_start_split()`, but
`_start_split()` only logs an at-capacity/stopped `_do_submit()` failure, so
`submit_scan_task()` still returns OK. This code then keeps the worker in
`remaining` even though it can never run or be retried, and the retained split
prevents dispatch destruction, so `multiget_data_v2` never calls `done`. Please
make admission atomic/observable and retire rejected splits, with a real
task-executor saturation/stop test.
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -312,104 +508,152 @@ 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 {
+
+// Both storage reads and scheduler 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;
}
-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());
+struct RowIdScanDispatch {
+ 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;
+
+ ~RowIdScanDispatch() {
+ // Scheduler 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 scan tasks discarded by
scheduler shutdown"));
+ complete();
+ }
}
- TabletSchema full_read_schema;
- for (const ColumnPB& column_pb : request_block_desc.column_descs()) {
- full_read_schema.append_column(TabletColumn(column_pb));
+
+ void complete() {
+ // Release request buffers while the rowid tracker is attached, even if
+ // the scheduler retains finished task closures for longer.
+ auto completion = std::move(on_complete);
+ completion(status.status());
+ run_task = {};
}
- 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());
+ void finish_worker() {
+ if (remaining.fetch_sub(1) == 1) {
+ 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);
- }
+} // namespace
- // 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 RowIdStorageReader::submit_internal_scan_tasks(ScannerScheduler*
scheduler, size_t task_count,
+ int concurrency,
+
std::function<Status(size_t)> run_task,
+
std::function<void(Status)> on_complete) {
+ std::shared_ptr<RowIdScanDispatch> dispatch;
+ auto setup_status = rowid_read_status([&]() {
+ dispatch = std::make_shared<RowIdScanDispatch>();
+ 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 scanner thread waiting for work submitted to its own
scheduler.
+ 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([&]() {
+ return scheduler->submit_scan_task(
+ SimplifiedScanTask(
+ [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 true;
+ },
+ nullptr, nullptr),
+ fmt::format("rowid-fetch-{}-{}", fmt::ptr(dispatch.get()),
i));
+ });
+ if (!status.ok()) {
Review Comment:
[P1] Do not retire this worker solely because `submit_scan_task()` threw. In
task-executor `enqueue_splits()`, `_start_split()` queues/wakes the split
before `finished_futures.push_back()`; that later allocation can throw.
`rowid_read_status` converts it to non-OK, this branch decrements `remaining`
as if nothing was admitted, and the submitter can then complete the BRPC and
clear `run_task` while the awakened worker is inside it, racing
request/`ReadRequestState` destruction. This is a new scheduler-handoff window,
distinct from the old bthread helper. Preallocate all post-admission state or
use a per-worker lifetime/once token, and test an injected throw after
admission.
--
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]
