Coverage Report

Created: 2026-08-21 05:47

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/fs/hdfs_file_writer.h
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
#pragma once
19
20
#include "common/status.h"
21
#include "io/fs/file_writer.h"
22
#include "io/fs/hdfs.h"
23
#include "io/fs/path.h"
24
25
namespace doris {
26
struct Slice;
27
namespace io {
28
29
class HdfsHandler;
30
class BlockFileCache;
31
struct FileCacheAllocatorBuilder;
32
struct AsyncCloseStatusPack;
33
34
Status validate_hdfs_write_batch_buffer_size(int64_t batch_buffer_size_mb,
35
                                             int64_t file_cache_block_size);
36
37
class HdfsFileWriter final : public FileWriter {
38
public:
39
    // Accepted path format:
40
    // - fs_name/path_to_file
41
    // - /path_to_file
42
    // TODO(plat1ko): Support related path for cloud mode
43
    static Result<FileWriterPtr> create(Path path, std::shared_ptr<HdfsHandler> handler,
44
                                        const std::string& fs_name,
45
                                        const FileWriterOptions* opts = nullptr);
46
47
    HdfsFileWriter(Path path, std::shared_ptr<HdfsHandler> handler, hdfsFile hdfs_file,
48
                   std::string fs_name, const FileWriterOptions* opts = nullptr);
49
    ~HdfsFileWriter() override;
50
51
    Status appendv(const Slice* data, size_t data_cnt) override;
52
0
    const Path& path() const override { return _path; }
53
1
    size_t bytes_appended() const override { return _bytes_appended; }
54
2
    State state() const override { return _state; }
55
56
    Status close(bool non_block = false) override;
57
58
private:
59
    Status _close_impl();
60
    // Flush buffered data into HDFS client and write local file cache if enabled
61
    // **Notice**: this would clear the underlying buffer
62
    Status _flush_buffer();
63
    Status append_hdfs_file(std::string_view content);
64
    void _write_into_local_file_cache();
65
    Status _append(std::string_view content);
66
    void _flush_and_reset_approximate_jni_buffer_size();
67
    Status _acquire_jni_memory(size_t size);
68
69
    Path _path;
70
    std::shared_ptr<HdfsHandler> _hdfs_handler = nullptr;
71
    hdfsFile _hdfs_file = nullptr;
72
    std::string _fs_name;
73
    size_t _bytes_appended = 0;
74
    bool _sync_file_data;
75
    class BatchBuffer {
76
    public:
77
        BatchBuffer(size_t capacity);
78
        size_t append(std::string_view content);
79
        bool full() const;
80
        const char* data() const;
81
        size_t capacity() const;
82
        size_t size() const;
83
        void clear();
84
        std::string_view content() const;
85
86
    private:
87
        std::string _batch_buffer;
88
    };
89
    BatchBuffer _batch_buffer;
90
    size_t _approximate_jni_buffer_size = 0;
91
    std::unique_ptr<AsyncCloseStatusPack> _async_close_pack;
92
    // We should make sure that close_impl's return value is consistent
93
    // So we need add one field to restore the value first time return by calling close_impl
94
    Status _st;
95
    State _state {State::OPENED};
96
};
97
98
} // namespace io
99
} // namespace doris