be/src/io/fs/hdfs_file_reader.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "io/fs/hdfs_file_reader.h"

#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"

namespace doris::io {

bvar::Adder<uint64_t> hdfs_bytes_read_total("hdfs_file_reader", "bytes_read");
bvar::LatencyRecorder hdfs_bytes_per_read("hdfs_file_reader", "bytes_per_read"); // also QPS
bvar::PerSecond<bvar::Adder<uint64_t>> hdfs_read_througthput("hdfs_file_reader",
                                                             "hdfs_read_throughput",
                                                             &hdfs_bytes_read_total);

namespace {

Result<FileHandleCache::Accessor> get_file(const hdfsFS& fs, const Path& file, int64_t mtime,
                                           int64_t file_size) {
    static FileHandleCache cache(config::max_hdfs_file_handle_cache_num, 16,
                                 config::max_hdfs_file_handle_cache_time_sec);
    bool cache_hit;
    FileHandleCache::Accessor accessor;
    RETURN_IF_ERROR_RESULT(cache.get_file_handle(fs, file.native(), mtime, file_size, false,
                                                 &accessor, &cache_hit));
    return accessor;
}

} // namespace

Result<FileReaderSPtr> HdfsFileReader::create(Path full_path, const hdfsFS& fs, std::string fs_name,
                                              const FileReaderOptions& opts,
                                              RuntimeProfile* profile) {
    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);
    });
}

HdfsFileReader::HdfsFileReader(Path path, std::string fs_name, FileHandleCache::Accessor accessor,
                               RuntimeProfile* profile, 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() {
    static_cast<void>(close());
}

Status HdfsFileReader::close() {
    bool expected = false;
    if (_closed.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) {
        DorisMetrics::instance()->hdfs_file_open_reading->increment(-1);
    }
    return Status::OK();
}

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;
        _accessor.destroy();
    }
    return st;
}

#ifdef USE_HADOOP_HDFS
Status HdfsFileReader::do_read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                                       const IOContext* /*io_ctx*/) {
    if (closed()) [[unlikely]] {
        return Status::InternalError("read closed file: {}", _path.native());
    }

    if (_handle == nullptr) [[unlikely]] {
        return Status::InternalError("cached hdfs file handle has been destroyed: {}",
                                     _path.native());
    }

    if (offset > _handle->file_size()) {
        return Status::IOError("offset exceeds file size(offset: {}, file size: {}, path: {})",
                               offset, _handle->file_size(), _path.native());
    }

    size_t bytes_req = result.size;
    char* to = result.data;
    bytes_req = std::min(bytes_req, (size_t)(_handle->file_size() - offset));
    *bytes_read = 0;
    if (UNLIKELY(bytes_req == 0)) {
        return Status::OK();
    }

    LIMIT_REMOTE_SCAN_IO(bytes_read);

    size_t has_read = 0;
    while (has_read < bytes_req) {
        int64_t max_to_read = bytes_req - has_read;
        tSize to_read = static_cast<tSize>(
                std::min(max_to_read, static_cast<int64_t>(std::numeric_limits<tSize>::max())));
        tSize loop_read = hdfsPread(_handle->fs(), _handle->file(), offset + has_read,
                                    to + has_read, to_read);
        {
            [[maybe_unused]] Status error_ret;
            TEST_INJECTION_POINT_RETURN_WITH_VALUE("HdfsFileReader:read_error", error_ret);
        }
        if (loop_read < 0) {
            // invoker maybe just skip Status.NotFound and continue
            // so we need distinguish between it and other kinds of errors
            std::string _err_msg = hdfs_error();
            if (_err_msg.find("No such file or directory") != std::string::npos) {
                return Status::NotFound(_err_msg);
            }
            return Status::InternalError(
                    "Read hdfs file failed. (BE: {}) namenode:{}, path:{}, err: {}",
                    BackendOptions::get_localhost(), _fs_name, _path.string(), _err_msg);
        }
        if (loop_read == 0) {
            break;
        }
        has_read += loop_read;
    }
    *bytes_read = has_read;
    hdfs_bytes_read_total << *bytes_read;
    hdfs_bytes_per_read << *bytes_read;
    return Status::OK();
}

#else
// The hedged read only support hdfsPread().
// TODO: rethink here to see if there are some difference between hdfsPread() and hdfsRead()
Status HdfsFileReader::do_read_at_impl(size_t offset, Slice result, size_t* bytes_read,
                                       const IOContext* /*io_ctx*/) {
    if (closed()) [[unlikely]] {
        return Status::InternalError("read closed file: ", _path.native());
    }

    if (offset > _handle->file_size()) {
        return Status::IOError("offset exceeds file size(offset: {}, file size: {}, path: {})",
                               offset, _handle->file_size(), _path.native());
    }

    int res = hdfsSeek(_handle->fs(), _handle->file(), offset);
    if (res != 0) {
        // invoker maybe just skip Status.NotFound and continue
        // so we need distinguish between it and other kinds of errors
        std::string _err_msg = hdfs_error();
        if (_err_msg.find("No such file or directory") != std::string::npos) {
            return Status::NotFound(_err_msg);
        }
        return Status::InternalError("Seek to offset failed. (BE: {}) offset={}, err: {}",
                                     BackendOptions::get_localhost(), offset, _err_msg);
    }

    size_t bytes_req = result.size;
    char* to = result.data;
    bytes_req = std::min(bytes_req, (size_t)(_handle->file_size() - offset));
    *bytes_read = 0;
    if (UNLIKELY(bytes_req == 0)) {
        return Status::OK();
    }

    LIMIT_REMOTE_SCAN_IO(bytes_read);

    size_t has_read = 0;
    while (has_read < bytes_req) {
        int64_t loop_read = hdfsRead(_handle->fs(), _handle->file(), to + has_read,
                                     static_cast<int32_t>(bytes_req - has_read));
        if (loop_read < 0) {
            // invoker maybe just skip Status.NotFound and continue
            // so we need distinguish between it and other kinds of errors
            std::string _err_msg = hdfs_error();
            if (_err_msg.find("No such file or directory") != std::string::npos) {
                return Status::NotFound(_err_msg);
            }
            return Status::InternalError(
                    "Read hdfs file failed. (BE: {}) namenode:{}, path:{}, err: {}",
                    BackendOptions::get_localhost(), _fs_name, _path.string(), _err_msg);
        }
        if (loop_read == 0) {
            break;
        }
        has_read += loop_read;
    }
    *bytes_read = has_read;
    hdfs_bytes_read_total << *bytes_read;
    hdfs_bytes_per_read << *bytes_read;
    return Status::OK();
}
#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

Coverage Report

Created: 2026-07-19 16:21

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "io/fs/hdfs_file_reader.h"
19
20		#include <stdint.h>
21
22		#include <algorithm>
23		#include <filesystem>
24		#include <ostream>
25		#include <utility>
26
27		#include "bvar/latency_recorder.h"
28		#include "bvar/reducer.h"
29		#include "common/compiler_util.h" // IWYU pragma: keep
30		#include "common/logging.h"
31		#include "common/metrics/doris_metrics.h"
32		#include "cpp/sync_point.h"
33		#include "io/fs/err_utils.h"
34		#include "io/hdfs_util.h"
35		#include "runtime/file_scan_profile.h"
36		#include "runtime/thread_context.h"
37		#include "runtime/workload_management/io_throttle.h"
38		#include "service/backend_options.h"
39
40		namespace doris::io {
41
42		bvar::Adder<uint64_t> hdfs_bytes_read_total("hdfs_file_reader", "bytes_read");
43		bvar::LatencyRecorder hdfs_bytes_per_read("hdfs_file_reader", "bytes_per_read"); // also QPS
44		bvar::PerSecond<bvar::Adder<uint64_t>> hdfs_read_througthput("hdfs_file_reader",
45		"hdfs_read_throughput",
46		&hdfs_bytes_read_total);
47
48		namespace {
49
50		Result<FileHandleCache::Accessor> get_file(const hdfsFS& fs, const Path& file, int64_t mtime,
51	0	int64_t file_size) {
52	0	static FileHandleCache cache(config::max_hdfs_file_handle_cache_num, 16,
53	0	config::max_hdfs_file_handle_cache_time_sec);
54	0	bool cache_hit;
55	0	FileHandleCache::Accessor accessor;
56	0	RETURN_IF_ERROR_RESULT(cache.get_file_handle(fs, file.native(), mtime, file_size, false,
57	0	&accessor, &cache_hit));
58	0	return accessor;
59	0	}
60
61		} // namespace
62
63		Result<FileReaderSPtr> HdfsFileReader::create(Path full_path, const hdfsFS& fs, std::string fs_name,
64		const FileReaderOptions& opts,
65	0	RuntimeProfile* profile) {
66	0	auto path = convert_path(full_path, fs_name);
67	0	return get_file(fs, path, opts.mtime, opts.file_size).transform([&](auto&& accessor) {
68	0	return std::make_shared<HdfsFileReader>(std::move(path), std::move(fs_name),
69	0	std::move(accessor), profile, opts.mtime);
70	0	});
71	0	}
72
73		HdfsFileReader::HdfsFileReader(Path path, std::string fs_name, FileHandleCache::Accessor accessor,
74		RuntimeProfile* profile, int64_t mtime)
75	0	: _path(std::move(path)),
76	0	_fs_name(std::move(fs_name)),
77	0	_accessor(std::move(accessor)),
78	0	_profile(profile),
79	0	_mtime(mtime) {
80	0	_handle = _accessor.get();
81
82	0	DorisMetrics::instance()->hdfs_file_open_reading->increment(1);
83	0	DorisMetrics::instance()->hdfs_file_reader_total->increment(1);
84	0	if (_profile != nullptr && is_hdfs(_fs_name)) {
85	0	#ifdef USE_HADOOP_HDFS
86	0	const char* hdfs_profile_name = "HdfsIO";
87	0	_total_read_time =
88	0	ADD_CHILD_TIMER(_profile, hdfs_profile_name,
89	0	file_scan_profile::parent_or_root(_profile, file_scan_profile::IO));
90	0	_hdfs_profile.total_bytes_read =
91	0	ADD_CHILD_COUNTER(_profile, "TotalBytesRead", TUnit::BYTES, hdfs_profile_name);
92	0	_hdfs_profile.total_local_bytes_read =
93	0	ADD_CHILD_COUNTER(_profile, "TotalLocalBytesRead", TUnit::BYTES, hdfs_profile_name);
94	0	_hdfs_profile.total_short_circuit_bytes_read = ADD_CHILD_COUNTER(
95	0	_profile, "TotalShortCircuitBytesRead", TUnit::BYTES, hdfs_profile_name);
96	0	_hdfs_profile.total_total_zero_copy_bytes_read = ADD_CHILD_COUNTER(
97	0	_profile, "TotalZeroCopyBytesRead", TUnit::BYTES, hdfs_profile_name);
98
99	0	_hdfs_profile.total_hedged_read =
100	0	ADD_CHILD_COUNTER(_profile, "TotalHedgedRead", TUnit::UNIT, hdfs_profile_name);
101	0	_hdfs_profile.hedged_read_in_cur_thread = ADD_CHILD_COUNTER(
102	0	_profile, "HedgedReadInCurThread", TUnit::UNIT, hdfs_profile_name);
103	0	_hdfs_profile.hedged_read_wins =
104	0	ADD_CHILD_COUNTER(_profile, "HedgedReadWins", TUnit::UNIT, hdfs_profile_name);
105	0	#endif
106	0	}
107	0	}
108
109	0	HdfsFileReader::~HdfsFileReader() {
110	0	static_cast<void>(close());
111	0	}
112
113	0	Status HdfsFileReader::close() {
114	0	bool expected = false;
115	0	if (_closed.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) {
116	0	DorisMetrics::instance()->hdfs_file_open_reading->increment(-1);
117	0	}
118	0	return Status::OK();
119	0	}
120
121		Status HdfsFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_read,
122	0	const IOContext* io_ctx) {
123	0	SCOPED_TIMER(_total_read_time);
124	0	auto st = do_read_at_impl(offset, result, bytes_read, io_ctx);
125	0	if (!st.ok()) {
126	0	_handle = nullptr;
127	0	_accessor.destroy();
128	0	}
129	0	return st;
130	0	}
131
132		#ifdef USE_HADOOP_HDFS
133		Status HdfsFileReader::do_read_at_impl(size_t offset, Slice result, size_t* bytes_read,
134	0	const IOContext* /io_ctx/) {
135	0	if (closed()) [[unlikely]] {
136	0	return Status::InternalError("read closed file: {}", _path.native());
137	0	}
138
139	0	if (_handle == nullptr) [[unlikely]] {
140	0	return Status::InternalError("cached hdfs file handle has been destroyed: {}",
141	0	_path.native());
142	0	}
143
144	0	if (offset > _handle->file_size()) {
145	0	return Status::IOError("offset exceeds file size(offset: {}, file size: {}, path: {})",
146	0	offset, _handle->file_size(), _path.native());
147	0	}
148
149	0	size_t bytes_req = result.size;
150	0	char* to = result.data;
151	0	bytes_req = std::min(bytes_req, (size_t)(_handle->file_size() - offset));
152	0	*bytes_read = 0;
153	0	if (UNLIKELY(bytes_req == 0)) {
154	0	return Status::OK();
155	0	}
156
157	0	LIMIT_REMOTE_SCAN_IO(bytes_read);
158
159	0	size_t has_read = 0;
160	0	while (has_read < bytes_req) {
161	0	int64_t max_to_read = bytes_req - has_read;
162	0	tSize to_read = static_cast<tSize>(
163	0	std::min(max_to_read, static_cast<int64_t>(std::numeric_limits<tSize>::max())));
164	0	tSize loop_read = hdfsPread(_handle->fs(), _handle->file(), offset + has_read,
165	0	to + has_read, to_read);
166	0	{
167	0	[[maybe_unused]] Status error_ret;
168	0	TEST_INJECTION_POINT_RETURN_WITH_VALUE("HdfsFileReader:read_error", error_ret);
169	0	}
170	0	if (loop_read < 0) {
171		// invoker maybe just skip Status.NotFound and continue
172		// so we need distinguish between it and other kinds of errors
173	0	std::string _err_msg = hdfs_error();
174	0	if (_err_msg.find("No such file or directory") != std::string::npos) {
175	0	return Status::NotFound(_err_msg);
176	0	}
177	0	return Status::InternalError(
178	0	"Read hdfs file failed. (BE: {}) namenode:{}, path:{}, err: {}",
179	0	BackendOptions::get_localhost(), _fs_name, _path.string(), _err_msg);
180	0	}
181	0	if (loop_read == 0) {
182	0	break;
183	0	}
184	0	has_read += loop_read;
185	0	}
186	0	*bytes_read = has_read;
187	0	hdfs_bytes_read_total << *bytes_read;
188	0	hdfs_bytes_per_read << *bytes_read;
189	0	return Status::OK();
190	0	}
191
192		#else
193		// The hedged read only support hdfsPread().
194		// TODO: rethink here to see if there are some difference between hdfsPread() and hdfsRead()
195		Status HdfsFileReader::do_read_at_impl(size_t offset, Slice result, size_t* bytes_read,
196		const IOContext* /io_ctx/) {
197		if (closed()) [[unlikely]] {
198		return Status::InternalError("read closed file: ", _path.native());
199		}
200
201		if (offset > _handle->file_size()) {
202		return Status::IOError("offset exceeds file size(offset: {}, file size: {}, path: {})",
203		offset, _handle->file_size(), _path.native());
204		}
205
206		int res = hdfsSeek(_handle->fs(), _handle->file(), offset);
207		if (res != 0) {
208		// invoker maybe just skip Status.NotFound and continue
209		// so we need distinguish between it and other kinds of errors
210		std::string _err_msg = hdfs_error();
211		if (_err_msg.find("No such file or directory") != std::string::npos) {
212		return Status::NotFound(_err_msg);
213		}
214		return Status::InternalError("Seek to offset failed. (BE: {}) offset={}, err: {}",
215		BackendOptions::get_localhost(), offset, _err_msg);
216		}
217
218		size_t bytes_req = result.size;
219		char* to = result.data;
220		bytes_req = std::min(bytes_req, (size_t)(_handle->file_size() - offset));
221		*bytes_read = 0;
222		if (UNLIKELY(bytes_req == 0)) {
223		return Status::OK();
224		}
225
226		LIMIT_REMOTE_SCAN_IO(bytes_read);
227
228		size_t has_read = 0;
229		while (has_read < bytes_req) {
230		int64_t loop_read = hdfsRead(_handle->fs(), _handle->file(), to + has_read,
231		static_cast<int32_t>(bytes_req - has_read));
232		if (loop_read < 0) {
233		// invoker maybe just skip Status.NotFound and continue
234		// so we need distinguish between it and other kinds of errors
235		std::string _err_msg = hdfs_error();
236		if (_err_msg.find("No such file or directory") != std::string::npos) {
237		return Status::NotFound(_err_msg);
238		}
239		return Status::InternalError(
240		"Read hdfs file failed. (BE: {}) namenode:{}, path:{}, err: {}",
241		BackendOptions::get_localhost(), _fs_name, _path.string(), _err_msg);
242		}
243		if (loop_read == 0) {
244		break;
245		}
246		has_read += loop_read;
247		}
248		*bytes_read = has_read;
249		hdfs_bytes_read_total << *bytes_read;
250		hdfs_bytes_per_read << *bytes_read;
251		return Status::OK();
252		}
253		#endif
254
255	0	void HdfsFileReader::_collect_profile_before_close() {
256	0	if (_profile != nullptr && is_hdfs(_fs_name)) {
257	0	#ifdef USE_HADOOP_HDFS
258	0	if (_handle == nullptr) [[unlikely]] {
259	0	return;
260	0	}
261
262	0	struct hdfsReadStatistics* hdfs_statistics = nullptr;
263	0	auto r = hdfsFileGetReadStatistics(_handle->file(), &hdfs_statistics);
264	0	if (r != 0) {
265	0	LOG(WARNING) << "Failed to run hdfsFileGetReadStatistics(): " << r
266	0	<< ", name node: " << _fs_name;
267	0	return;
268	0	}
269	0	COUNTER_UPDATE(_hdfs_profile.total_bytes_read, hdfs_statistics->totalBytesRead);
270	0	COUNTER_UPDATE(_hdfs_profile.total_local_bytes_read, hdfs_statistics->totalLocalBytesRead);
271	0	COUNTER_UPDATE(_hdfs_profile.total_short_circuit_bytes_read,
272	0	hdfs_statistics->totalShortCircuitBytesRead);
273	0	COUNTER_UPDATE(_hdfs_profile.total_total_zero_copy_bytes_read,
274	0	hdfs_statistics->totalZeroCopyBytesRead);
275	0	hdfsFileFreeReadStatistics(hdfs_statistics);
276
277	0	struct hdfsHedgedReadMetrics* hdfs_hedged_read_statistics = nullptr;
278	0	r = hdfsGetHedgedReadMetrics(_handle->fs(), &hdfs_hedged_read_statistics);
279	0	if (r != 0) {
280	0	LOG(WARNING) << "Failed to run hdfsGetHedgedReadMetrics(): " << r
281	0	<< ", name node: " << _fs_name;
282	0	return;
283	0	}
284
285	0	COUNTER_UPDATE(_hdfs_profile.total_hedged_read, hdfs_hedged_read_statistics->hedgedReadOps);
286	0	COUNTER_UPDATE(_hdfs_profile.hedged_read_in_cur_thread,
287	0	hdfs_hedged_read_statistics->hedgedReadOpsInCurThread);
288	0	COUNTER_UPDATE(_hdfs_profile.hedged_read_wins,
289	0	hdfs_hedged_read_statistics->hedgedReadOpsWin);
290
291	0	hdfsFreeHedgedReadMetrics(hdfs_hedged_read_statistics);
292	0	hdfsFileClearReadStatistics(_handle->file());
293	0	#endif
294	0	}
295	0	}
296
297		} // namespace doris::io