github-actions[bot] commented on code in PR #64372:
URL: https://github.com/apache/doris/pull/64372#discussion_r3819425113
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -855,14 +876,91 @@ Status
RowIdStorageReader::read_external_row_from_file_mapping(
file_read_times_counter->value(),
file_read_times_counter->type());
}
- semaphore.release();
- if (++producer_count == scan_rows_count) {
- std::lock_guard<std::mutex> lock(mtx);
- cv.notify_one();
- }
return Status::OK();
}
+std::string RowIdStorageReader::source_column_key(const SlotDescriptor& slot,
uint32_t column_idx) {
+ fmt::memory_buffer key;
+ // Length-prefix each component so distinct sequences cannot alias, e.g.
+ // paths ["a", "b"] -> "1:a1:b" while ["a:b"] -> "3:a:b".
+ auto append = [&key](std::string_view component) {
+ fmt::format_to(key, "{}:", component.size());
+ key.append(component.data(), component.data() + component.size());
+ };
+ append(slot.col_name());
+ append(std::to_string(column_idx));
+ append(std::to_string(slot.col_unique_id()));
+ append(std::to_string(slot.column_paths().size()));
+ for (const auto& path : slot.column_paths()) {
+ append(path);
+ }
+ append(std::to_string(slot.all_access_paths().size()));
+ // Encode each optional sub-path's presence bit separately from its element
+ // count so an absent path ("0") never aliases a present-but-empty path
+ // ("1" + size "0").
+ auto append_optional_path = [&append](bool is_set, const
std::vector<std::string>& items) {
+ append(is_set ? "1" : "0");
+ if (is_set) {
+ append(std::to_string(items.size()));
+ for (const auto& item : items) {
+ append(item);
+ }
+ }
+ };
+ for (const auto& path : slot.all_access_paths()) {
+ append(fmt::format("{}", path.type));
+ append_optional_path(path.__isset.data_access_path,
path.data_access_path.path);
+ append_optional_path(path.__isset.meta_access_path,
path.meta_access_path.path);
+ }
+ return fmt::to_string(key);
+}
+
+Status RowIdStorageReader::submit_external_scan_tasks(
+ ScannerScheduler* scheduler, std::counting_semaphore<>& semaphore,
size_t task_count,
+ const std::function<std::string(size_t)>& make_task_id,
+ const std::function<Status(size_t)>& run_task) {
+ // `completed_count` is a plain counter guarded by `mtx`; the same mutex
guards
+ // the wait predicate below, so a worker can never notify between the
waiter's
+ // predicate check and its wait.
+ AtomicStatus scan_status;
+ std::condition_variable cv;
+ std::mutex mtx;
+ size_t completed_count = 0;
+
+ // Only tasks the scheduler actually accepted are waited for. If a
submission
+ // fails we stop submitting, but still wait for the already-accepted tasks
so
+ // their workers cannot outlive the locals they capture by reference.
+ size_t submitted_count = 0;
+ for (size_t idx = 0; idx < task_count; ++idx) {
+ semaphore.acquire();
+ Status submit_st =
+ scheduler->submit_scan_task(SimplifiedScanTask(
+ [&, idx]() -> bool {
+ Defer complete([&] {
+
std::lock_guard<std::mutex> lock(mtx);
+ ++completed_count;
+ cv.notify_one();
+ });
+
scan_status.update(run_task(idx));
Review Comment:
[P1] Publish worker exceptions before signaling completion
If `run_task(idx)` throws, this call never reaches `AtomicStatus::update`,
but the completion defer still increments `completed_count` and wakes the
waiter (and the inner scan defer releases its permit). The task executor
catches `doris::Exception` only into the split's private future, which this
helper never observes, so the RPC path can proceed with an OK `scan_status`,
lose the original reader error, and skip the scheduler wrapper's task-removal
path. Catch and convert the repository-standard exception set inside this
callback, publish it to `scan_status`, and still return `true`; a
throwing-worker unit test should cover this path.
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -993,42 +1119,69 @@ Status RowIdStorageReader::read_batch_external_row(
int64_t scan_running_time = 0;
RETURN_IF_ERROR(scope_timer_run(
[&]() -> Status {
- // Make sure to insert data into result_block only after all
scan tasks have been executed.
- std::atomic<int> producer_count {0};
- std::condition_variable cv;
- std::mutex mtx;
-
//semaphore: Limit the number of scan tasks submitted at one
time
std::counting_semaphore semaphore {max_file_scanners};
- size_t idx = 0;
+ std::vector<std::pair<std::multimap<segment_v2::rowid_t,
size_t>,
+ std::shared_ptr<FileMapping>>>
+ scan_info_list;
+ scan_info_list.reserve(scan_rows.size());
for (const auto& [_, scan_info] : scan_rows) {
- semaphore.acquire();
- RETURN_IF_ERROR(remote_scan_sched->submit_scan_task(
- SimplifiedScanTask(
- [&, idx, scan_info]() -> Status {
- const auto& [row_ids, file_mapping] =
scan_info;
- return
read_external_row_from_file_mapping(
- idx, row_ids, file_mapping,
slots, query_id,
- runtime_state, scan_blocks,
row_id_block_idx,
- fetch_statistics,
rpc_scan_params,
- colname_to_slot_id,
producer_count,
- scan_rows.size(), semaphore,
cv, mtx, tuple_desc);
- },
- nullptr, nullptr),
- fmt::format("{}-read_batch_external_row-{}",
print_id(query_id), idx)));
- idx++;
+ scan_info_list.emplace_back(scan_info);
}
- {
- std::unique_lock<std::mutex> lock(mtx);
- cv.wait(lock, [&] { return producer_count ==
scan_rows.size(); });
- }
- return Status::OK();
+ return submit_external_scan_tasks(
+ remote_scan_sched, semaphore, scan_rows.size(),
+ [&](size_t idx) {
+ return
fmt::format("{}-read_batch_external_row-{}", print_id(query_id),
+ idx);
+ },
+ [&](size_t idx) -> Status {
+ const auto& [row_ids, file_mapping] =
scan_info_list[idx];
+ return read_external_row_from_file_mapping(
+ idx, row_ids, file_mapping, scan_slots,
query_id, runtime_state,
+ scan_blocks, row_id_block_idx,
fetch_statistics,
+ rpc_scan_params, colname_to_slot_id,
semaphore, tuple_desc);
+ });
},
&scan_running_time));
- scatter_scan_blocks_to_result_block(row_id_block_idx, scan_blocks,
result_block);
+ // Insert the read data into result_block. Use insert_indices_from()
instead of
+ // scatter_scan_blocks_to_result_block()/insert_from_multi_column(),
because
+ // scan_blocks may have fewer columns than result_block when duplicate
physical columns
+ // are deduplicated, and insert_from_multi_column() cannot handle
ColumnString
+ // cross-type (32/64) copies safely.
+ uint32_t scan_position = 0;
+ for (size_t column_id = 0; column_id < result_block.get_columns().size();
column_id++) {
Review Comment:
[P2] Use constant-time column counts in the scatter loop
`Block::get_columns()` allocates a vector and calls
`convert_to_full_column_if_const()` for every column. Calling it in this loop
condition makes a C-column projection do O(C^2) column visits, and the inner
DCHECK repeats the same full traversal per output cell in checked builds. These
checks only need the count, so use `result_block.columns()` here and
`scan_blocks[pos_block].columns()` below.
##########
regression-test/suites/external_table_p0/hive/test_hive_topn_lazy_mat.groovy:
##########
@@ -34,6 +34,15 @@ suite("test_hive_topn_lazy_mat", "p0,external") {
qt_3 """ select score, value, active,name from ${table} order by
id,file_id limit 10; """
qt_4 """ select value,name,id,file_id from ${table} order by name
limit 10; """
+ // Duplicate projected column - same column twice (the core bug
from rowid_fetcher fix)
+ qt_dup_col_twice """ select name a, name b from ${table} order by
id limit 10; """
Review Comment:
[P2] Prove this duplicate projection exercises phase 2
This result is identical when the suite later disables
`topn_lazy_materialization_threshold`, and the existing EXPLAIN assertions
cover different projection shapes. If duplicate aliases make this exact query
fall back to eager scanning, the new regression stays green without entering
the changed row-id dedup/remap path. Add an enabled-path EXPLAIN for this
duplicate query (and the analogous TVF shape) that requires `VMaterializeNode`,
the row-id slot, and the expected duplicate descriptor/index layout.
--
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]
