Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions be/src/agent/task_worker_pool.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
9 changes: 4 additions & 5 deletions be/src/cloud/cloud_storage_engine.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion be/src/exec/sink/writer/vfile_result_writer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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: {
Expand Down
4 changes: 2 additions & 2 deletions be/src/io/file_factory.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
});
Expand Down
6 changes: 3 additions & 3 deletions be/src/io/fs/benchmark/hdfs_benchmark.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
Expand All @@ -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));
Expand All @@ -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;
Expand Down
77 changes: 3 additions & 74 deletions be/src/io/fs/hdfs_file_reader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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() {
Expand All @@ -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;
Expand Down Expand Up @@ -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
24 changes: 2 additions & 22 deletions be/src/io/fs/hdfs_file_reader.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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
17 changes: 7 additions & 10 deletions be/src/io/fs/hdfs_file_system.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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;
}
Expand Down Expand Up @@ -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();
}

Expand Down
7 changes: 2 additions & 5 deletions be/src/io/fs/hdfs_file_system.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -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
2 changes: 1 addition & 1 deletion be/src/runtime/snapshot_loader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading