diff --git a/be/src/agent/task_worker_pool.cpp b/be/src/agent/task_worker_pool.cpp index 2605112673c8c1..694a58a9674c09 100644 --- a/be/src/agent/task_worker_pool.cpp +++ b/be/src/agent/task_worker_pool.cpp @@ -1747,9 +1747,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 { diff --git a/be/src/cloud/cloud_storage_engine.cpp b/be/src/cloud/cloud_storage_engine.cpp index 91d096fc47cd75..1727b44720c50f 100644 --- a/be/src/cloud/cloud_storage_engine.cpp +++ b/be/src/cloud/cloud_storage_engine.cpp @@ -153,8 +153,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(); @@ -183,9 +183,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(hdfs_fs); put_storage_resource(id, {std::move(hdfs), path_format}, 0); return Status::OK(); diff --git a/be/src/exec/sink/writer/vfile_result_writer.cpp b/be/src/exec/sink/writer/vfile_result_writer.cpp index c198eaa7e21f4e..089ac2828bbf42 100644 --- a/be/src/exec/sink/writer/vfile_result_writer.cpp +++ b/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: { diff --git a/be/src/io/file_factory.cpp b/be/src/io/file_factory.cpp index 92fb19ff5d4932..7bbbc4a5cd020e 100644 --- a/be/src/io/file_factory.cpp +++ b/be/src/io/file_factory.cpp @@ -129,7 +129,7 @@ Result 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; @@ -255,7 +255,7 @@ Result 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); }); diff --git a/be/src/io/fs/benchmark/hdfs_benchmark.hpp b/be/src/io/fs/benchmark/hdfs_benchmark.hpp index 5b3161e6397ce6..6e23c914153b0a 100644 --- a/be/src/io/fs/benchmark/hdfs_benchmark.hpp +++ b/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; diff --git a/be/src/io/fs/hdfs_file_reader.cpp b/be/src/io/fs/hdfs_file_reader.cpp index 505a2da1976343..db8214c12bbd9e 100644 --- a/be/src/io/fs/hdfs_file_reader.cpp +++ b/be/src/io/fs/hdfs_file_reader.cpp @@ -20,19 +20,15 @@ #include #include -#include -#include #include #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_management/io_throttle.h" #include "service/backend_options.h" @@ -62,49 +58,24 @@ Result get_file(const hdfsFS& fs, const Path& file, i } // namespace Result 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(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() { @@ -121,7 +92,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; @@ -253,47 +223,6 @@ 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 - } -} #include "common/compile_check_end.h" } // namespace doris::io diff --git a/be/src/io/fs/hdfs_file_reader.h b/be/src/io/fs/hdfs_file_reader.h index 7bb73f30909452..b18b8d8b5433b6 100644 --- a/be/src/io/fs/hdfs_file_reader.h +++ b/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 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 _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 diff --git a/be/src/io/fs/hdfs_file_system.cpp b/be/src/io/fs/hdfs_file_system.cpp index a137a2934a05c0..e85b582e24f9a8 100644 --- a/be/src/io/fs/hdfs_file_system.cpp +++ b/be/src/io/fs/hdfs_file_system.cpp @@ -54,14 +54,13 @@ namespace doris::io { Result> HdfsFileSystem::create( const std::map& 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> 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> HdfsFileSystem::create(const THdfsParams "true.")); } #endif - std::shared_ptr fs(new HdfsFileSystem( - hdfs_params, std::move(fs_name), std::move(id), profile, std::move(root_path))); + std::shared_ptr 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(); } diff --git a/be/src/io/fs/hdfs_file_system.h b/be/src/io/fs/hdfs_file_system.h index 0b71f636a87bb6..6190302c1b68bd 100644 --- a/be/src/io/fs/hdfs_file_system.h +++ b/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> create(const THdfsParams& hdfs_params, std::string fs_name, std::string id, - RuntimeProfile* profile, std::string root_path = ""); static Result> create( const std::map& 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 _fs_handler = nullptr; - RuntimeProfile* _profile = nullptr; }; } // namespace io } // namespace doris diff --git a/be/src/runtime/snapshot_loader.cpp b/be/src/runtime/snapshot_loader.cpp index 6ffcc598d45f7d..720f332e595d12 100644 --- a/be/src/runtime/snapshot_loader.cpp +++ b/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 fs; _remote_fs = DORIS_TRY(