Coverage Report

Created: 2026-09-14 19:09

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