From ec36f8feb79e9fae400566432b609ec094044df0 Mon Sep 17 00:00:00 2001 From: morningman Date: Sun, 25 Oct 2026 14:54:20 +0800 Subject: [PATCH] [fix](be) Correct the external row fetch InitReaderAvgTime and GetBlockAvgTime ### What problem does this PR solve? Issue Number: None Related PR: #67005 Problem Summary: The second phase of TopN lazy materialization on an external table (RowIdStorageReader::read_batch_external_row) reads each scan range of a request block with its own reader and reports two profile values per request block: InitReaderAvgTime and GetBlockAvgTime, the mean reader-init and block-read time over those ranges. MaterializationOperator lists them per backend in the query profile. Both are wrong on master: - InitReaderAvgTime assigns each range's init time instead of adding it, so it reports the last range's time divided by the number of ranges: ranges that took 10, 20 and 60 ms give 20 ms instead of 30 ms. - GetBlockAvgTime is filled with the init-time average, so the block-read time never reaches the profile. - The averages are accumulated in the caller's variables, which read_by_rowids() zeroes once per RPC and passes to every request block. When an RPC fetches two lazily materialized external tables, the second block's sums start from the first block's averages. branch-4.1/4.2 fixed the first two in #67005 (a Lance profile PR) by summing the init times and reporting the get-block average, but kept the shared accumulators, so a second request block reports inflated averages there: 46ms and 466ms instead of 30ms and 300ms in the unit test below once the first block left 50 and 500 ms. The summary now lives in RowIdStorageReader::add_external_fetch_avg_time_profile(). It sums both times over the block's ranges from zero, overwrites the caller's variables with the averages and reports each average under its own key. The "Query stats" log line at the end of read_by_rowids() therefore prints the averages of the RPC's last external request block, the same block whose scan range count it already prints. ### Release note None ### Check List (For Author) - Test: Unit Test - BE UT RowIdStorageReaderTest.ExternalFetchProfileAveragesInitAndGetBlockTimes: three ranges (10/100, 20/200, 60/600 ms) with the caller's variables preset to 50/500 ms by a previous request block give 30ms and 300ms. - Behavior changed: No - Does this need documentation: No Co-authored-by: zhangstar333 Co-Authored-By: Claude Opus 5.5 (1M context) --- be/src/exec/rowid_fetcher.cpp | 31 ++++++++++++++++++++-------- be/src/exec/rowid_fetcher.h | 11 ++++++++++ be/test/exec/rowid_fetcher_test.cpp | 32 +++++++++++++++++++++++++++++ 3 files changed, 66 insertions(+), 8 deletions(-) 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({});