[fix](hdfs) Remove query profile from HDFS file reader

### What problem does this PR solve?

Issue Number: None

Related PR: apache#67293

Problem Summary: `CachedRemoteFileReader::prefetch_range()` submits fire-and-forget dry-run tasks that can outlive the query `RuntimeProfile`. `HdfsFileReader` stored raw pointers to query-profile timers and counters, so a delayed prefetch read could update them after the query profile had been destroyed and trigger a use-after-free. Remove the query `RuntimeProfile` dependency and HDFS-specific query-profile timers and counters from `HdfsFileReader`, remove the unused profile passthrough from `HdfsFileSystem`, and update all affected call sites. Generic file-reader and file-cache statistics plus process-wide HDFS bvars remain unchanged.

### Release note

None

### Check List (For Author)

- Test: Manual test
    - `build-support/clang-format.sh`
    - `build-support/check-format.sh`
    - `build-support/check-build-hygiene.sh`
    - `git diff --check upstream/master...HEAD`
- Behavior changed: Yes. Query profiles no longer expose the HDFS-specific `HdfsIO` timer and HDFS read-statistics counters; HDFS reads and file-cache behavior are unchanged.
- Does this need documentation: No

(cherry picked from commit 4e45489)

Author: bobhan1

be/src/agent/task_worker_pool.cpp

@@ -1750,9 +1750,9 @@ void update_hdfs_resource(const TStorageResource& param, io::RemoteFileSystemSPt
 
     if (!existed_fs) {
         // No such FS instance on BE
-        auto res = io::HdfsFileSystem::create(
-                param.hdfs_storage_param, param.hdfs_storage_param.fs_name,
-                std::to_string(param.id), nullptr, std::move(root_path));
+        auto res = io::HdfsFileSystem::create(param.hdfs_storage_param,
+                                              param.hdfs_storage_param.fs_name,
+                                              std::to_string(param.id), std::move(root_path));
         if (!res.has_value()) {
             st = std::move(res).error();
         } else {

be/src/cloud/cloud_storage_engine.cpp

@@ -166,8 +166,8 @@ struct VaultCreateFSVisitor {
     // TODO(ByteYue): Make sure enable_java_support is on
     Status operator()(const cloud::HdfsVaultInfo& vault) const {
         auto hdfs_params = io::to_hdfs_params(vault);
-        auto fs = DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name, id,
-                                                       nullptr, vault.prefix()));
+        auto fs = DORIS_TRY(
+                io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name, id, vault.prefix()));
         put_storage_resource(id, {std::move(fs), path_format}, 0);
         LOG_INFO("successfully create hdfs vault, vault id {}", id);
         return Status::OK();
@@ -196,9 +196,8 @@ struct RefreshFSVaultVisitor {
 
     Status operator()(const cloud::HdfsVaultInfo& vault) const {
         auto hdfs_params = io::to_hdfs_params(vault);
-        auto hdfs_fs =
-                DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name, id, nullptr,
-                                                     vault.has_prefix() ? vault.prefix() : ""));
+        auto hdfs_fs = DORIS_TRY(io::HdfsFileSystem::create(
+                hdfs_params, hdfs_params.fs_name, id, vault.has_prefix() ? vault.prefix() : ""));
         auto hdfs = std::static_pointer_cast<io::HdfsFileSystem>(hdfs_fs);
         put_storage_resource(id, {std::move(hdfs), path_format}, 0);
         return Status::OK();

be/src/exec/sink/writer/vfile_result_writer.cpp

@@ -447,7 +447,7 @@ Status VFileResultWriter::_delete_dir() {
     case TStorageBackendType::HDFS: {
         THdfsParams hdfs_params = parse_properties(_file_opts->broker_properties);
         auto fs = DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name,
-                                                       io::FileSystem::TMP_FS_ID, nullptr));
+                                                       io::FileSystem::TMP_FS_ID));
         return fs->delete_directory(dir);
     }
     case TStorageBackendType::S3: {

be/src/io/file_factory.cpp

@@ -123,7 +123,7 @@ Result<io::FileSystemSPtr> FileFactory::create_fs(const io::FSPropertiesRef& fs_
     case TFileType::FILE_HDFS: {
         std::string fs_name = _get_fs_name(file_description);
         return io::HdfsFileSystem::create(*fs_properties.properties, fs_name,
-                                          io::FileSystem::TMP_FS_ID, nullptr);
+                                          io::FileSystem::TMP_FS_ID);
     }
     case TFileType::FILE_HTTP: {
         const auto& kv = *fs_properties.properties;
@@ -249,7 +249,7 @@ Result<io::FileReaderSPtr> FileFactory::_create_file_reader_internal(
         RETURN_IF_ERROR_RESULT(ExecEnv::GetInstance()->hdfs_mgr()->get_or_create_fs(
                 system_properties.hdfs_params, *fs_name, &handler));
         return io::HdfsFileReader::create(file_description.path, handler->hdfs_fs, *fs_name,
-                                          reader_options, profile)
+                                          reader_options)
                 .and_then([&](auto&& reader) {
                     return io::create_cached_file_reader(std::move(reader), reader_options);
                 });

be/src/io/fs/benchmark/hdfs_benchmark.hpp

@@ -95,7 +95,7 @@ class HdfsCreateWriteBenchmark : public BaseBenchmark {
         io::FileWriterPtr writer;
         THdfsParams hdfs_params = parse_properties(_conf_map);
         auto fs = DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name,
-                                                       io::FileSystem::TMP_FS_ID, nullptr));
+                                                       io::FileSystem::TMP_FS_ID));
         RETURN_IF_ERROR(fs->create_file(file_path, &writer));
         return write(state, writer.get());
     }
@@ -116,7 +116,7 @@ class HdfsRenameBenchmark : public BaseBenchmark {
         auto new_file_path = file_path + "_new";
         THdfsParams hdfs_params = parse_properties(_conf_map);
         auto fs = DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name,
-                                                       io::FileSystem::TMP_FS_ID, nullptr));
+                                                       io::FileSystem::TMP_FS_ID));
 
         auto start = std::chrono::high_resolution_clock::now();
         RETURN_IF_ERROR(fs->rename(file_path, new_file_path));
@@ -143,7 +143,7 @@ class HdfsExistsBenchmark : public BaseBenchmark {
 
         THdfsParams hdfs_params = parse_properties(_conf_map);
         auto fs = DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name,
-                                                       io::FileSystem::TMP_FS_ID, nullptr));
+                                                       io::FileSystem::TMP_FS_ID));
 
         auto start = std::chrono::high_resolution_clock::now();
         bool res = false;

be/src/io/fs/hdfs_file_reader.cpp

@@ -20,19 +20,15 @@
 #include <stdint.h>
 
 #include <algorithm>
-#include <filesystem>
-#include <ostream>
 #include <utility>
 
 #include "bvar/latency_recorder.h"
 #include "bvar/reducer.h"
 #include "common/compiler_util.h" // IWYU pragma: keep
-#include "common/logging.h"
 #include "common/metrics/doris_metrics.h"
 #include "cpp/sync_point.h"
 #include "io/fs/err_utils.h"
 #include "io/hdfs_util.h"
-#include "runtime/file_scan_profile.h"
 #include "runtime/thread_context.h"
 #include "runtime/workload_group/workload_group.h"
 #include "runtime/workload_management/io_throttle.h"
@@ -63,49 +59,24 @@ Result<FileHandleCache::Accessor> get_file(const hdfsFS& fs, const Path& file, i
 } // namespace
 
 Result<FileReaderSPtr> HdfsFileReader::create(Path full_path, const hdfsFS& fs, std::string fs_name,
-                                              const FileReaderOptions& opts,
-                                              RuntimeProfile* profile) {
+                                              const FileReaderOptions& opts) {
     auto path = convert_path(full_path, fs_name);
     return get_file(fs, path, opts.mtime, opts.file_size).transform([&](auto&& accessor) {
         return std::make_shared<HdfsFileReader>(std::move(path), std::move(fs_name),
-                                                std::move(accessor), profile, opts.mtime);
+                                                std::move(accessor), opts.mtime);
     });
 }
 
 HdfsFileReader::HdfsFileReader(Path path, std::string fs_name, FileHandleCache::Accessor accessor,
-                               RuntimeProfile* profile, int64_t mtime)
+                               int64_t mtime)
         : _path(std::move(path)),
           _fs_name(std::move(fs_name)),
           _accessor(std::move(accessor)),
-          _profile(profile),
           _mtime(mtime) {
     _handle = _accessor.get();
 
     DorisMetrics::instance()->hdfs_file_open_reading->increment(1);
     DorisMetrics::instance()->hdfs_file_reader_total->increment(1);
-    if (_profile != nullptr && is_hdfs(_fs_name)) {
-#ifdef USE_HADOOP_HDFS
-        const char* hdfs_profile_name = "HdfsIO";
-        _total_read_time =
-                ADD_CHILD_TIMER(_profile, hdfs_profile_name,
-                                file_scan_profile::parent_or_root(_profile, file_scan_profile::IO));
-        _hdfs_profile.total_bytes_read =
-                ADD_CHILD_COUNTER(_profile, "TotalBytesRead", TUnit::BYTES, hdfs_profile_name);
-        _hdfs_profile.total_local_bytes_read =
-                ADD_CHILD_COUNTER(_profile, "TotalLocalBytesRead", TUnit::BYTES, hdfs_profile_name);
-        _hdfs_profile.total_short_circuit_bytes_read = ADD_CHILD_COUNTER(
-                _profile, "TotalShortCircuitBytesRead", TUnit::BYTES, hdfs_profile_name);
-        _hdfs_profile.total_total_zero_copy_bytes_read = ADD_CHILD_COUNTER(
-                _profile, "TotalZeroCopyBytesRead", TUnit::BYTES, hdfs_profile_name);
-
-        _hdfs_profile.total_hedged_read =
-                ADD_CHILD_COUNTER(_profile, "TotalHedgedRead", TUnit::UNIT, hdfs_profile_name);
-        _hdfs_profile.hedged_read_in_cur_thread = ADD_CHILD_COUNTER(
-                _profile, "HedgedReadInCurThread", TUnit::UNIT, hdfs_profile_name);
-        _hdfs_profile.hedged_read_wins =
-                ADD_CHILD_COUNTER(_profile, "HedgedReadWins", TUnit::UNIT, hdfs_profile_name);
-#endif
-    }
 }
 
 HdfsFileReader::~HdfsFileReader() {
@@ -122,7 +93,6 @@ Status HdfsFileReader::close() {
 
 Status HdfsFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                                     const IOContext* io_ctx) {
-    SCOPED_TIMER(_total_read_time);
     auto st = do_read_at_impl(offset, result, bytes_read, io_ctx);
     if (!st.ok()) {
         _handle = nullptr;
@@ -254,46 +224,4 @@ Status HdfsFileReader::do_read_at_impl(size_t offset, Slice result, size_t* byte
 }
 #endif
 
-void HdfsFileReader::_collect_profile_before_close() {
-    if (_profile != nullptr && is_hdfs(_fs_name)) {
-#ifdef USE_HADOOP_HDFS
-        if (_handle == nullptr) [[unlikely]] {
-            return;
-        }
-
-        struct hdfsReadStatistics* hdfs_statistics = nullptr;
-        auto r = hdfsFileGetReadStatistics(_handle->file(), &hdfs_statistics);
-        if (r != 0) {
-            LOG(WARNING) << "Failed to run hdfsFileGetReadStatistics(): " << r
-                         << ", name node: " << _fs_name;
-            return;
-        }
-        COUNTER_UPDATE(_hdfs_profile.total_bytes_read, hdfs_statistics->totalBytesRead);
-        COUNTER_UPDATE(_hdfs_profile.total_local_bytes_read, hdfs_statistics->totalLocalBytesRead);
-        COUNTER_UPDATE(_hdfs_profile.total_short_circuit_bytes_read,
-                       hdfs_statistics->totalShortCircuitBytesRead);
-        COUNTER_UPDATE(_hdfs_profile.total_total_zero_copy_bytes_read,
-                       hdfs_statistics->totalZeroCopyBytesRead);
-        hdfsFileFreeReadStatistics(hdfs_statistics);
-
-        struct hdfsHedgedReadMetrics* hdfs_hedged_read_statistics = nullptr;
-        r = hdfsGetHedgedReadMetrics(_handle->fs(), &hdfs_hedged_read_statistics);
-        if (r != 0) {
-            LOG(WARNING) << "Failed to run hdfsGetHedgedReadMetrics(): " << r
-                         << ", name node: " << _fs_name;
-            return;
-        }
-
-        COUNTER_UPDATE(_hdfs_profile.total_hedged_read, hdfs_hedged_read_statistics->hedgedReadOps);
-        COUNTER_UPDATE(_hdfs_profile.hedged_read_in_cur_thread,
-                       hdfs_hedged_read_statistics->hedgedReadOpsInCurThread);
-        COUNTER_UPDATE(_hdfs_profile.hedged_read_wins,
-                       hdfs_hedged_read_statistics->hedgedReadOpsWin);
-
-        hdfsFreeHedgedReadMetrics(hdfs_hedged_read_statistics);
-        hdfsFileClearReadStatistics(_handle->file());
-#endif
-    }
-}
-
 } // namespace doris::io

be/src/io/fs/hdfs_file_reader.h

@@ -42,10 +42,10 @@ class HdfsFileReader final : public FileReader {
     // - /path_to_file
     // TODO(plat1ko): Support related path for cloud mode
     static Result<FileReaderSPtr> create(Path path, const hdfsFS& fs, std::string fs_name,
-                                         const FileReaderOptions& opts, RuntimeProfile* profile);
+                                         const FileReaderOptions& opts);
 
     HdfsFileReader(Path path, std::string fs_name, FileHandleCache::Accessor accessor,
-                   RuntimeProfile* profile, int64_t mtime = 0);
+                   int64_t mtime = 0);
 
     ~HdfsFileReader() override;
 
@@ -63,35 +63,15 @@ class HdfsFileReader final : public FileReader {
     Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                         const IOContext* io_ctx) override;
 
-    void _collect_profile_before_close() override;
-
     Status do_read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                            const IOContext* io_ctx);
 
 private:
-#ifdef USE_HADOOP_HDFS
-    struct HDFSProfile {
-        RuntimeProfile::Counter* total_bytes_read = nullptr;
-        RuntimeProfile::Counter* total_local_bytes_read = nullptr;
-        RuntimeProfile::Counter* total_short_circuit_bytes_read = nullptr;
-        RuntimeProfile::Counter* total_total_zero_copy_bytes_read = nullptr;
-
-        RuntimeProfile::Counter* total_hedged_read = nullptr;
-        RuntimeProfile::Counter* hedged_read_in_cur_thread = nullptr;
-        RuntimeProfile::Counter* hedged_read_wins = nullptr;
-    };
-#endif
-
     Path _path;
     std::string _fs_name;
     FileHandleCache::Accessor _accessor;
     CachedHdfsFileHandle* _handle = nullptr; // owned by _cached_file_handle
     std::atomic<bool> _closed = false;
-    RuntimeProfile* _profile = nullptr;
-    RuntimeProfile::Counter* _total_read_time = nullptr;
     int64_t _mtime;
-#ifdef USE_HADOOP_HDFS
-    HDFSProfile _hdfs_profile;
-#endif
 };
 } // namespace doris::io

be/src/io/fs/hdfs_file_system.cpp

@@ -54,14 +54,13 @@ namespace doris::io {
 
 Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(
         const std::map<std::string, std::string>& properties, std::string fs_name, std::string id,
-        RuntimeProfile* profile, std::string root_path) {
+        std::string root_path) {
     return HdfsFileSystem::create(parse_properties(properties), std::move(fs_name), std::move(id),
-                                  profile, std::move(root_path));
+                                  std::move(root_path));
 }
 
 Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(const THdfsParams& hdfs_params,
                                                                std::string fs_name, std::string id,
-                                                               RuntimeProfile* profile,
                                                                std::string root_path) {
 #ifdef USE_HADOOP_HDFS
     if (!config::enable_java_support) {
@@ -70,18 +69,17 @@ Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(const THdfsParams
                 "true."));
     }
 #endif
-    std::shared_ptr<HdfsFileSystem> fs(new HdfsFileSystem(
-            hdfs_params, std::move(fs_name), std::move(id), profile, std::move(root_path)));
+    std::shared_ptr<HdfsFileSystem> fs(new HdfsFileSystem(hdfs_params, std::move(fs_name),
+                                                          std::move(id), std::move(root_path)));
     RETURN_IF_ERROR_RESULT(fs->init());
     return fs;
 }
 
 HdfsFileSystem::HdfsFileSystem(const THdfsParams& hdfs_params, std::string fs_name, std::string id,
-                               RuntimeProfile* profile, std::string root_path)
+                               std::string root_path)
         : RemoteFileSystem(std::move(root_path), std::move(id), FileSystemType::HDFS),
           _hdfs_params(hdfs_params),
-          _fs_name(std::move(fs_name)),
-          _profile(profile) {
+          _fs_name(std::move(fs_name)) {
     if (_fs_name.empty()) {
         _fs_name = hdfs_params.fs_name;
     }
@@ -112,8 +110,7 @@ Status HdfsFileSystem::create_file_impl(const Path& file, FileWriterPtr* writer,
 Status HdfsFileSystem::open_file_internal(const Path& file, FileReaderSPtr* reader,
                                           const FileReaderOptions& opts) {
     CHECK_HDFS_HANDLER(_fs_handler);
-    *reader =
-            DORIS_TRY(HdfsFileReader::create(file, _fs_handler->hdfs_fs, _fs_name, opts, _profile));
+    *reader = DORIS_TRY(HdfsFileReader::create(file, _fs_handler->hdfs_fs, _fs_name, opts));
     return Status::OK();
 }
 

be/src/io/fs/hdfs_file_system.h

@@ -32,7 +32,6 @@
 #include "io/fs/hdfs.h"
 #include "io/fs/path.h"
 #include "io/fs/remote_file_system.h"
-#include "runtime/runtime_profile.h"
 
 namespace doris {
 class THdfsParams;
@@ -48,12 +47,11 @@ class HdfsFileSystem final : public RemoteFileSystem {
 public:
     static Result<std::shared_ptr<HdfsFileSystem>> create(const THdfsParams& hdfs_params,
                                                           std::string fs_name, std::string id,
-                                                          RuntimeProfile* profile,
                                                           std::string root_path = "");
 
     static Result<std::shared_ptr<HdfsFileSystem>> create(
             const std::map<std::string, std::string>& properties, std::string fs_name,
-            std::string id, RuntimeProfile* profile, std::string root_path = "");
+            std::string id, std::string root_path = "");
 
     ~HdfsFileSystem() override;
 
@@ -85,11 +83,10 @@ class HdfsFileSystem final : public RemoteFileSystem {
 private:
     friend class HdfsFileWriter;
     HdfsFileSystem(const THdfsParams& hdfs_params, std::string fs_name, std::string id,
-                   RuntimeProfile* profile, std::string root_path);
+                   std::string root_path);
     const THdfsParams& _hdfs_params; // Only used in init, so we can use reference here
     std::string _fs_name;
     std::shared_ptr<HdfsHandler> _fs_handler = nullptr;
-    RuntimeProfile* _profile = nullptr;
 };
 } // namespace io
 } // namespace doris

be/src/runtime/snapshot_loader.cpp

@@ -766,7 +766,7 @@ Status BaseSnapshotLoader::init(TStorageBackendType::type type, const std::strin
     } else if (TStorageBackendType::type::HDFS == type) {
         THdfsParams hdfs_params = parse_properties(_prop);
         _remote_fs = DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name,
-                                                          io::FileSystem::TMP_FS_ID, nullptr));
+                                                          io::FileSystem::TMP_FS_ID));
     } else if (TStorageBackendType::type::BROKER == type) {
         std::shared_ptr<io::BrokerFileSystem> fs;
         _remote_fs = DORIS_TRY(