[fix](be) Correct the external row fetch InitReaderAvgTime and GetBlockAvgTime

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<ExternalFetchStatistics>& 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<int64_t>(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<IdFileMap> id_file_map, std::vector<SlotDescriptor>& 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,

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<std::string(size_t)>& make_task_id,
                                              const std::function<Status(size_t)>& 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<ExternalFetchStatistics>& 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;

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<RowIdStorageReader::ExternalFetchStatistics> 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({});