diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp index 3f9c0efc82ebd9..615b553990bfba 100644 --- a/be/src/exec/rowid_fetcher.cpp +++ b/be/src/exec/rowid_fetcher.cpp @@ -685,6 +685,27 @@ TFileScanRangeParams RowIdStorageReader::build_external_scan_params( return params; } +void RowIdStorageReader::add_external_fetch_avg_time_profile( + const std::vector& fetch_statistics, + RuntimeProfile* runtime_profile, int64_t* init_reader_avg_ms, int64_t* get_block_avg_ms) { + // read_by_rowids() reads a request block only when it has a row ID, and every row ID lands + // in a scan range, so there is at least one range to average over. + DORIS_CHECK(!fetch_statistics.empty()); + int64_t init_reader_total_ms = 0; + int64_t get_block_total_ms = 0; + for (const auto& statistics : fetch_statistics) { + init_reader_total_ms += statistics.init_reader_ms; + get_block_total_ms += statistics.get_block_ms; + } + const auto range_count = static_cast(fetch_statistics.size()); + *init_reader_avg_ms = init_reader_total_ms / range_count; + *get_block_avg_ms = get_block_total_ms / range_count; + runtime_profile->add_info_string(InitReaderAvgTimeProfile, + std::to_string(*init_reader_avg_ms) + "ms"); + runtime_profile->add_info_string(GetBlockAvgTimeProfile, + std::to_string(*get_block_avg_ms) + "ms"); +} + Status RowIdStorageReader::read_batch_external_row( const uint64_t workload_group_id, const PRequestBlockDesc& request_block_desc, std::shared_ptr id_file_map, std::vector& slots, @@ -902,8 +923,6 @@ Status RowIdStorageReader::read_batch_external_row( size_t idx = 0; for (const auto& [_, scan_info] : scan_rows) { format_to(file_read_lines_buffer, "{}, ", scan_info.first.size()); - *init_reader_avg_ms = fetch_statistics[idx].init_reader_ms; - *get_block_avg_ms += fetch_statistics[idx].get_block_ms; format_to(file_read_bytes_buffer, "{}, ", fetch_statistics[idx].file_read_bytes); format_to(file_read_times_buffer, "{}, ", fetch_statistics[idx].file_read_times); idx++; @@ -913,12 +932,8 @@ Status RowIdStorageReader::read_batch_external_row( format_to(file_read_bytes_buffer, "]"); format_to(file_read_times_buffer, "]"); - *init_reader_avg_ms /= fetch_statistics.size(); - *get_block_avg_ms /= fetch_statistics.size(); - runtime_profile->add_info_string(InitReaderAvgTimeProfile, - std::to_string(*init_reader_avg_ms) + "ms"); - runtime_profile->add_info_string(GetBlockAvgTimeProfile, - std::to_string(*init_reader_avg_ms) + "ms"); + add_external_fetch_avg_time_profile(fetch_statistics, runtime_profile.get(), + init_reader_avg_ms, get_block_avg_ms); runtime_profile->add_info_string(FileReadLinesProfile, fmt::to_string(file_read_lines_buffer)); runtime_profile->add_info_string(FileScannerV2::FileReadBytesProfile, diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h index 916b77ff214a62..ad06a12d993a81 100644 --- a/be/src/exec/rowid_fetcher.h +++ b/be/src/exec/rowid_fetcher.h @@ -35,6 +35,7 @@ namespace doris { +class RuntimeProfile; class RuntimeState; class TQueryOptions; class TupleDescriptor; @@ -148,6 +149,16 @@ class RowIdStorageReader { const std::function& make_task_id, const std::function& run_task); + // Reports InitReaderAvgTime and GetBlockAvgTime of one request block: the mean reader-init + // and block-read times over its scan ranges, e.g. 20ms and 200ms for two ranges that took + // 10/100 ms and 30/300 ms. Both outputs are overwritten rather than accumulated, because an + // RPC that fetches two lazily materialized tables reads its two request blocks into the same + // caller variables. + static void add_external_fetch_avg_time_profile( + const std::vector& fetch_statistics, + RuntimeProfile* runtime_profile, int64_t* init_reader_avg_ms, + int64_t* get_block_avg_ms); + struct ExternalFetchStatistics { int64_t init_reader_ms = 0; int64_t get_block_ms = 0; diff --git a/be/test/exec/rowid_fetcher_test.cpp b/be/test/exec/rowid_fetcher_test.cpp index b0478cfc4abd74..3c6189a849eb02 100644 --- a/be/test/exec/rowid_fetcher_test.cpp +++ b/be/test/exec/rowid_fetcher_test.cpp @@ -33,6 +33,7 @@ #include "runtime/descriptor_helper.h" #include "runtime/descriptors.h" #include "runtime/exec_env.h" +#include "runtime/runtime_profile.h" #include "runtime/runtime_state.h" #include "storage/id_manager.h" #include "storage/rowset_id.h" @@ -301,6 +302,37 @@ TEST_F(RowIdStorageReaderTest, ExternalFetchPartitionSlotsPreserveHivePositionMa } } +// A request block reads each of its scan ranges with a separate reader, and its profile reports the +// mean reader-init and block-read time over those ranges. When an RPC fetches two lazily +// materialized tables, the second request block starts with the first block's averages in the +// caller's variables; they must not leak into its own. +TEST_F(RowIdStorageReaderTest, ExternalFetchProfileAveragesInitAndGetBlockTimes) { + std::vector fetch_statistics(3); + fetch_statistics[0].init_reader_ms = 10; + fetch_statistics[0].get_block_ms = 100; + fetch_statistics[1].init_reader_ms = 20; + fetch_statistics[1].get_block_ms = 200; + fetch_statistics[2].init_reader_ms = 60; + fetch_statistics[2].get_block_ms = 600; + // The averages a previous request block of the same RPC left behind. + int64_t init_reader_avg_ms = 50; + int64_t get_block_avg_ms = 500; + RuntimeProfile profile("ExternalRowIDFetcher"); + + RowIdStorageReader::add_external_fetch_avg_time_profile(fetch_statistics, &profile, + &init_reader_avg_ms, &get_block_avg_ms); + + EXPECT_EQ(init_reader_avg_ms, 30); + EXPECT_EQ(get_block_avg_ms, 300); + const auto* init_reader_avg = + profile.get_info_string(RowIdStorageReader::InitReaderAvgTimeProfile); + ASSERT_NE(init_reader_avg, nullptr); + EXPECT_EQ(*init_reader_avg, "30ms"); + const auto* get_block_avg = profile.get_info_string(RowIdStorageReader::GetBlockAvgTimeProfile); + ASSERT_NE(get_block_avg, nullptr); + EXPECT_EQ(*get_block_avg, "300ms"); +} + TEST_F(RowIdStorageReaderTest, SameSourceColumnSharesKey) { // The bug case: one physical column projected twice must dedup onto one scan column. const SlotDescriptor first = make_slot({});