This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new d084ff1efb0 branch-4.1: [fix](hdfs) Remove query profile from HDFS
file reader (#67293)
d084ff1efb0 is described below
commit d084ff1efb0476bab1a793d093057afab3253409
Author: bobhan1 <[email protected]>
AuthorDate: Mon Aug 31 18:16:06 2026 +0800
branch-4.1: [fix](hdfs) Remove query profile from HDFS file reader (#67293)
### What problem does this PR solve?
pick https://github.com/apache/doris/pull/67335
Issue Number: DORIS-28268
Related PR: None
Problem Summary:
`CachedRemoteFileReader::prefetch_range()` submits fire-and-forget
dry-run tasks. A running task keeps the cached reader and its underlying
`HdfsFileReader` alive, but it does not keep the query `RuntimeProfile`
alive. Because `HdfsFileReader` stored raw pointers to query-profile
timers and counters, 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/counters from `HdfsFileReader`. This also removes
the now-unused profile passthrough from `HdfsFileSystem` and updates all
affected call sites. Generic file-reader/file-cache statistics and
process-wide HDFS bvars remain unchanged; HDFS reads and cache behavior
are unchanged.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [ ] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- `clang-format --dry-run --Werror` on all changed C++ files
- `git diff --check`
- `./build.sh --be -j100`: the changed HDFS reader/filesystem and
affected call-site translation units compiled successfully. The full
local BE build is blocked by an unrelated installed Lance C header
mismatch (`LanceScanStatistics` and statistics callback definitions are
missing from the local `lance.h`).
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes. Query profiles no longer expose the HDFS-specific `HdfsIO`
timer and HDFS read-statistics counters. HDFS read and file-cache
behavior are unchanged.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/agent/task_worker_pool.cpp | 6 +-
be/src/cloud/cloud_storage_engine.cpp | 9 ++-
be/src/exec/sink/writer/vfile_result_writer.cpp | 2 +-
be/src/io/file_factory.cpp | 4 +-
be/src/io/fs/benchmark/hdfs_benchmark.hpp | 6 +-
be/src/io/fs/hdfs_file_reader.cpp | 77 +------------------------
be/src/io/fs/hdfs_file_reader.h | 24 +-------
be/src/io/fs/hdfs_file_system.cpp | 17 +++---
be/src/io/fs/hdfs_file_system.h | 7 +--
be/src/runtime/snapshot_loader.cpp | 2 +-
10 files changed, 28 insertions(+), 126 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 6ee1554a874..cf124f67656 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 91d096fc47c..1727b44720c 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<io::HdfsFileSystem>(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 c198eaa7e21..089ac2828bb 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 92fb19ff5d4..7bbbc4a5cd0 100644
--- a/be/src/io/file_factory.cpp
+++ b/be/src/io/file_factory.cpp
@@ -129,7 +129,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;
@@ -255,7 +255,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);
});
diff --git a/be/src/io/fs/benchmark/hdfs_benchmark.hpp
b/be/src/io/fs/benchmark/hdfs_benchmark.hpp
index 5b3161e6397..6e23c914153 100644
--- a/be/src/io/fs/benchmark/hdfs_benchmark.hpp
+++ b/be/src/io/fs/benchmark/hdfs_benchmark.hpp
@@ -95,7 +95,7 @@ public:
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 @@ public:
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 @@ public:
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 505a2da1976..db8214c12bb 100644
--- a/be/src/io/fs/hdfs_file_reader.cpp
+++ b/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_management/io_throttle.h"
#include "service/backend_options.h"
@@ -62,49 +58,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() {
@@ -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 7bb73f30909..b18b8d8b543 100644
--- a/be/src/io/fs/hdfs_file_reader.h
+++ b/be/src/io/fs/hdfs_file_reader.h
@@ -42,10 +42,10 @@ public:
// - /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 @@ protected:
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
diff --git a/be/src/io/fs/hdfs_file_system.cpp
b/be/src/io/fs/hdfs_file_system.cpp
index a137a2934a0..e85b582e24f 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<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();
}
diff --git a/be/src/io/fs/hdfs_file_system.h b/be/src/io/fs/hdfs_file_system.h
index 0b71f636a87..6190302c1b6 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<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 @@ private:
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
diff --git a/be/src/runtime/snapshot_loader.cpp
b/be/src/runtime/snapshot_loader.cpp
index 6ffcc598d45..720f332e595 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<io::BrokerFileSystem> fs;
_remote_fs = DORIS_TRY(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
