Coverage Report

Created: 2026-08-21 02:18

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/fs/hdfs_file_system.cpp
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_system.h"
19
20
#include <errno.h>
21
#include <fcntl.h>
22
#include <gen_cpp/PlanNodes_types.h>
23
24
#include <algorithm>
25
#include <filesystem>
26
#include <map>
27
#include <mutex>
28
#include <ostream>
29
#include <unordered_map>
30
#include <utility>
31
32
#include "common/config.h"
33
#include "common/status.h"
34
#include "core/pod_array.h"
35
#include "io/fs/err_utils.h"
36
#include "io/fs/hdfs/hdfs_mgr.h"
37
#include "io/fs/hdfs_file_reader.h"
38
#include "io/fs/hdfs_file_writer.h"
39
#include "io/fs/local_file_system.h"
40
#include "io/hdfs_builder.h"
41
#include "io/hdfs_util.h"
42
#include "runtime/exec_env.h"
43
#include "util/bvar_helper.h"
44
#include "util/obj_lru_cache.h"
45
#include "util/slice.h"
46
47
namespace doris::io {
48
49
#ifndef CHECK_HDFS_HANDLER
50
#define CHECK_HDFS_HANDLER(handler)                        \
51
0
    if (!handler) {                                        \
52
0
        return Status::IOError("init Hdfs handler error"); \
53
0
    }
54
#endif
55
56
Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(
57
        const std::map<std::string, std::string>& properties, std::string fs_name, std::string id,
58
0
        RuntimeProfile* profile, std::string root_path) {
59
0
    return HdfsFileSystem::create(parse_properties(properties), std::move(fs_name), std::move(id),
60
0
                                  profile, std::move(root_path));
61
0
}
62
63
Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(const THdfsParams& hdfs_params,
64
                                                               std::string fs_name, std::string id,
65
                                                               RuntimeProfile* profile,
66
0
                                                               std::string root_path) {
67
0
#ifdef USE_HADOOP_HDFS
68
0
    if (!config::enable_java_support) {
69
0
        return ResultError(Status::InternalError(
70
0
                "hdfs file system is not enabled, you can change be config enable_java_support to "
71
0
                "true."));
72
0
    }
73
0
#endif
74
0
    std::shared_ptr<HdfsFileSystem> fs(new HdfsFileSystem(
75
0
            hdfs_params, std::move(fs_name), std::move(id), profile, std::move(root_path)));
76
0
    RETURN_IF_ERROR_RESULT(fs->init());
77
0
    return fs;
78
0
}
79
80
HdfsFileSystem::HdfsFileSystem(const THdfsParams& hdfs_params, std::string fs_name, std::string id,
81
                               RuntimeProfile* profile, std::string root_path)
82
0
        : RemoteFileSystem(std::move(root_path), std::move(id), FileSystemType::HDFS),
83
0
          _hdfs_params(hdfs_params),
84
0
          _fs_name(std::move(fs_name)),
85
0
          _profile(profile) {
86
0
    if (_fs_name.empty()) {
87
0
        _fs_name = hdfs_params.fs_name;
88
0
    }
89
0
}
90
91
0
HdfsFileSystem::~HdfsFileSystem() = default;
92
93
0
Status HdfsFileSystem::init() {
94
0
    RETURN_IF_ERROR(ExecEnv::GetInstance()->hdfs_mgr()->get_or_create_fs(_hdfs_params, _fs_name,
95
0
                                                                         &_fs_handler));
96
0
    if (!_fs_handler) {
97
0
        return Status::InternalError("failed to init Hdfs handler with, please check hdfs params.");
98
0
    }
99
0
    return Status::OK();
100
0
}
101
102
Status HdfsFileSystem::create_file_impl(const Path& file, FileWriterPtr* writer,
103
0
                                        const FileWriterOptions* opts) {
104
0
    auto res = io::HdfsFileWriter::create(file, _fs_handler, _fs_name, opts);
105
0
    if (res.has_value()) {
106
0
        *writer = std::move(res).value();
107
0
        return Status::OK();
108
0
    } else {
109
0
        return std::move(res).error();
110
0
    }
111
0
}
112
113
Status HdfsFileSystem::open_file_internal(const Path& file, FileReaderSPtr* reader,
114
0
                                          const FileReaderOptions& opts) {
115
0
    CHECK_HDFS_HANDLER(_fs_handler);
116
0
    *reader =
117
0
            DORIS_TRY(HdfsFileReader::create(file, _fs_handler->hdfs_fs, _fs_name, opts, _profile));
118
0
    return Status::OK();
119
0
}
120
121
0
Status HdfsFileSystem::create_directory_impl(const Path& dir, bool failed_if_exists) {
122
0
    CHECK_HDFS_HANDLER(_fs_handler);
123
0
    Path real_path = convert_path(dir, _fs_name);
124
0
    int res;
125
0
    {
126
0
        SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_create_dir_latency);
127
0
        res = hdfsCreateDirectory(_fs_handler->hdfs_fs, real_path.string().c_str());
128
0
    }
129
0
    if (res == -1) {
130
0
        return Status::IOError("failed to create directory {}: {}", dir.native(), hdfs_error());
131
0
    }
132
0
    return Status::OK();
133
0
}
134
135
0
Status HdfsFileSystem::delete_file_impl(const Path& file) {
136
0
    return delete_internal(file, 0);
137
0
}
138
139
0
Status HdfsFileSystem::delete_directory_impl(const Path& dir) {
140
0
    return delete_internal(dir, 1);
141
0
}
142
143
0
Status HdfsFileSystem::batch_delete_impl(const std::vector<Path>& files) {
144
0
    for (auto& file : files) {
145
0
        RETURN_IF_ERROR(delete_file_impl(file));
146
0
    }
147
0
    return Status::OK();
148
0
}
149
150
0
Status HdfsFileSystem::delete_internal(const Path& path, int is_recursive) {
151
0
    bool exists = true;
152
0
    RETURN_IF_ERROR(exists_impl(path, &exists));
153
0
    if (!exists) {
154
0
        return Status::OK();
155
0
    }
156
0
    CHECK_HDFS_HANDLER(_fs_handler);
157
0
    Path real_path = convert_path(path, _fs_name);
158
0
    int res = hdfsDelete(_fs_handler->hdfs_fs, real_path.string().c_str(), is_recursive);
159
0
    if (res == -1) {
160
0
        return Status::IOError("failed to delete directory {}: {}", path.native(), hdfs_error());
161
0
    }
162
0
    return Status::OK();
163
0
}
164
165
0
Status HdfsFileSystem::exists_impl(const Path& path, bool* res) const {
166
0
    CHECK_HDFS_HANDLER(_fs_handler);
167
0
    Path real_path = convert_path(path, _fs_name);
168
0
    int is_exists = hdfsExists(_fs_handler->hdfs_fs, real_path.string().c_str());
169
0
#ifdef USE_HADOOP_HDFS
170
    // when calling hdfsExists() and return non-zero code,
171
    // if errno is ENOENT, which means the file does not exist.
172
    // if errno is not ENOENT, which means it encounter other error, should return.
173
    // NOTE: not for libhdfs3 since it only runs on MaxOS, don't have to support it.
174
    //
175
    // See details:
176
    //  https://github.com/apache/hadoop/blob/5cda162a804fb0cfc2a5ac0058ab407662c5fb00/
177
    //  hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/hdfs.c#L1923-L1924
178
0
    if (is_exists != 0 && errno != ENOENT) {
179
0
        char* root_cause = hdfsGetLastExceptionRootCause();
180
0
        return Status::IOError("failed to check path existence {}: {}", path.native(),
181
0
                               (root_cause ? root_cause : "unknown"));
182
0
    }
183
0
#endif
184
0
    *res = (is_exists == 0);
185
0
    return Status::OK();
186
0
}
187
188
0
Status HdfsFileSystem::file_size_impl(const Path& path, int64_t* file_size) const {
189
0
    CHECK_HDFS_HANDLER(_fs_handler);
190
0
    Path real_path = convert_path(path, _fs_name);
191
0
    SCOPED_BVAR_LATENCY(hdfs_bvar::hdfs_get_path_info_latency);
192
0
    hdfsFileInfo* file_info = hdfsGetPathInfo(_fs_handler->hdfs_fs, real_path.string().c_str());
193
0
    if (file_info == nullptr) {
194
0
        return Status::IOError("failed to get file size of {}: {}", path.native(), hdfs_error());
195
0
    }
196
0
    *file_size = file_info->mSize;
197
0
    hdfsFreeFileInfo(file_info, 1);
198
0
    return Status::OK();
199
0
}
200
201
Status HdfsFileSystem::list_impl(const Path& path, bool only_file, std::vector<FileInfo>* files,
202
0
                                 bool* exists) {
203
0
    RETURN_IF_ERROR(exists_impl(path, exists));
204
0
    if (!(*exists)) {
205
0
        return Status::OK();
206
0
    }
207
208
0
    CHECK_HDFS_HANDLER(_fs_handler);
209
0
    Path real_path = convert_path(path, _fs_name);
210
0
    int numEntries = 0;
211
0
    hdfsFileInfo* hdfs_file_info =
212
0
            hdfsListDirectory(_fs_handler->hdfs_fs, real_path.c_str(), &numEntries);
213
0
    if (hdfs_file_info == nullptr) {
214
0
        return Status::IOError("failed to list files/directors {}: {}", path.native(),
215
0
                               hdfs_error());
216
0
    }
217
0
    for (int idx = 0; idx < numEntries; ++idx) {
218
0
        auto& file = hdfs_file_info[idx];
219
0
        if (only_file && file.mKind == kObjectKindDirectory) {
220
0
            continue;
221
0
        }
222
0
        auto& file_info = files->emplace_back();
223
0
        std::string_view fname(file.mName);
224
0
        fname.remove_prefix(fname.rfind('/') + 1);
225
0
        file_info.file_name = fname;
226
0
        file_info.file_size = file.mSize;
227
0
        file_info.is_file = (file.mKind != kObjectKindDirectory);
228
0
    }
229
0
    hdfsFreeFileInfo(hdfs_file_info, numEntries);
230
0
    return Status::OK();
231
0
}
232
233
0
Status HdfsFileSystem::rename_impl(const Path& orig_name, const Path& new_name) {
234
0
    Path normal_orig_name = convert_path(orig_name, _fs_name);
235
0
    Path normal_new_name = convert_path(new_name, _fs_name);
236
0
    int ret = hdfsRename(_fs_handler->hdfs_fs, normal_orig_name.c_str(), normal_new_name.c_str());
237
0
    if (ret == 0) {
238
0
        LOG(INFO) << "finished to rename file. orig: " << normal_orig_name
239
0
                  << ", new: " << normal_new_name;
240
0
        return Status::OK();
241
0
    } else {
242
0
        return Status::IOError("fail to rename from {} to {}: {}", normal_orig_name.native(),
243
0
                               normal_new_name.native(), hdfs_error());
244
0
    }
245
0
    return Status::OK();
246
0
}
247
248
0
Status HdfsFileSystem::upload_impl(const Path& local_file, const Path& remote_file) {
249
    // 1. open local file for read
250
0
    FileSystemSPtr local_fs = global_local_filesystem();
251
0
    FileReaderSPtr local_reader = nullptr;
252
0
    RETURN_IF_ERROR(local_fs->open_file(local_file, &local_reader));
253
0
    int64_t file_len = local_reader->size();
254
0
    if (file_len == -1) {
255
0
        return Status::IOError("failed to get size of file: {}", local_file.string());
256
0
    }
257
258
    // 2. open remote file for write
259
0
    FileWriterPtr hdfs_writer = nullptr;
260
0
    RETURN_IF_ERROR(create_file_impl(remote_file, &hdfs_writer, nullptr));
261
262
0
    constexpr size_t buf_sz = 1024 * 1024;
263
0
    char read_buf[buf_sz];
264
0
    size_t left_len = file_len;
265
0
    size_t read_offset = 0;
266
0
    size_t bytes_read = 0;
267
0
    while (left_len > 0) {
268
0
        size_t read_len = left_len > buf_sz ? buf_sz : left_len;
269
0
        RETURN_IF_ERROR(local_reader->read_at(read_offset, {read_buf, read_len}, &bytes_read));
270
0
        RETURN_IF_ERROR(hdfs_writer->append({read_buf, read_len}));
271
272
0
        read_offset += read_len;
273
0
        left_len -= read_len;
274
0
    }
275
276
0
    return hdfs_writer->close();
277
0
}
278
279
Status HdfsFileSystem::batch_upload_impl(const std::vector<Path>& local_files,
280
0
                                         const std::vector<Path>& remote_files) {
281
0
    DCHECK(local_files.size() == remote_files.size());
282
0
    for (int i = 0; i < local_files.size(); ++i) {
283
0
        RETURN_IF_ERROR(upload_impl(local_files[i], remote_files[i]));
284
0
    }
285
0
    return Status::OK();
286
0
}
287
288
0
Status HdfsFileSystem::download_impl(const Path& remote_file, const Path& local_file) {
289
    // 1. open remote file for read
290
0
    FileReaderSPtr hdfs_reader = nullptr;
291
0
    RETURN_IF_ERROR(open_file_internal(remote_file, &hdfs_reader, FileReaderOptions::DEFAULT));
292
293
    // 2. remove the existing local file if exist
294
0
    if (std::filesystem::remove(local_file)) {
295
0
        LOG(INFO) << "remove the previously exist local file: " << local_file;
296
0
    }
297
298
    // 3. open local file for write
299
0
    FileSystemSPtr local_fs = global_local_filesystem();
300
0
    FileWriterPtr local_writer = nullptr;
301
0
    RETURN_IF_ERROR(local_fs->create_file(local_file, &local_writer));
302
303
    // 4. read remote and write to local
304
0
    LOG(INFO) << "read remote file: " << remote_file << " to local: " << local_file;
305
0
    constexpr size_t buf_sz = 1024 * 1024;
306
0
    PODArray<char> read_buf;
307
0
    read_buf.resize(buf_sz);
308
0
    size_t cur_offset = 0;
309
0
    while (true) {
310
0
        size_t read_len = 0;
311
0
        Slice file_slice(read_buf.data(), buf_sz);
312
0
        RETURN_IF_ERROR(hdfs_reader->read_at(cur_offset, file_slice, &read_len));
313
0
        cur_offset += read_len;
314
0
        if (read_len == 0) {
315
0
            break;
316
0
        }
317
318
0
        RETURN_IF_ERROR(local_writer->append({read_buf.data(), read_len}));
319
0
    }
320
0
    return local_writer->close();
321
0
}
322
323
} // namespace doris::io