Coverage Report

Created: 2026-10-23 15:33

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/fs/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 <future>
21
#include <memory>
22
23
#include "common/status.h"
24
#include "io/cache/block_file_cache.h"
25
#include "io/cache/block_file_cache_factory.h"
26
#include "io/cache/file_cache_common.h"
27
#include "io/fs/file_reader_writer_fwd.h"
28
#include "io/fs/packed_slice_location.h"
29
#include "io/fs/path.h"
30
#include "util/slice.h"
31
32
namespace doris::io {
33
class FileSystem;
34
struct FileCacheAllocatorBuilder;
35
struct EncryptionInfo;
36
37
// Only affects remote file writers
38
struct FileWriterOptions {
39
    // BE stages every file as an uncompleted upload (S3 multipart parts or Azure blocks), and
40
    // the FE coordinator completes them at commit. An uncompleted file is not visible, so a
41
    // single file is atomic. Multiple files are still not atomic, but because Hive committers
42
    // have best-effort semantics, this shortens the inconsistent time window.
43
    // The coordinator must support the client's ObjStorageCapabilities::deferred_completion.
44
    bool deferred_completion = false;
45
    bool write_file_cache = false;
46
    bool allow_adaptive_file_cache_write = true;
47
    bool is_cold_data = false;
48
    bool sync_file_data = true;              // Whether flush data into storage system
49
    uint64_t file_cache_expiration_time = 0; // Absolute time, 0 means no TTL
50
    uint64_t approximate_bytes_to_write = 0; // Approximate bytes to write, used for file cache
51
};
52
53
struct AsyncCloseStatusPack {
54
    std::promise<Status> promise;
55
    std::future<Status> future;
56
};
57
58
class FileWriter {
59
public:
60
    enum class State : uint8_t {
61
        OPENED = 0,
62
        ASYNC_CLOSING,
63
        CLOSED,
64
    };
65
371k
    FileWriter() = default;
66
372k
    virtual ~FileWriter() = default;
67
68
    FileWriter(const FileWriter&) = delete;
69
    const FileWriter& operator=(const FileWriter&) = delete;
70
71
    // Normal close. Wait for all data to persist before returning.
72
    // If there is no data appended, an empty file will be persisted.
73
    virtual Status close(bool non_block = false) = 0;
74
75
    // Non-blocking probe for a previous close(true).
76
    // OK means close finished successfully. NeedSendAgain means close is still running.
77
    // Other errors mean close finished with error or the writer does not support this API.
78
    // NOTE: This method consumes the async close result when it is ready. The caller must
79
    // use it as the only completion path for that async close; mixing it with close(false)
80
    // or another try_finish_close consumer is not supported.
81
0
    virtual Status try_finish_close() {
82
0
        return Status::NotSupported("try_finish_close is not supported");
83
0
    }
84
85
50.7k
    Status append(const Slice& data) { return appendv(&data, 1); }
86
87
    virtual Status appendv(const Slice* data, size_t data_cnt) = 0;
88
89
    virtual const Path& path() const = 0;
90
91
    virtual size_t bytes_appended() const = 0;
92
93
    virtual State state() const = 0;
94
95
    // Gets the location of this file's slice in the packed file. Must be called after the writer
96
    // is closed. An empty packed_file_path means the file is not in a packed file. Writers that
97
    // wrap another writer must forward this call, so callers never need to downcast to
98
    // PackedFileWriter.
99
1
    virtual Status get_packed_slice_location(PackedSliceLocation* location) const {
100
1
        *location = PackedSliceLocation {};
101
1
        return Status::OK();
102
1
    }
103
104
    FileCacheAllocatorBuilder* cache_builder() const {
105
        return _cache_builder == nullptr ? nullptr : _cache_builder.get();
106
    }
107
108
protected:
109
76.8k
    void init_cache_builder(const FileWriterOptions* opts, const Path& path) {
110
76.9k
        if (!config::enable_file_cache || opts == nullptr) {
111
1.71k
            return;
112
1.71k
        }
113
114
75.0k
        io::UInt128Wrapper path_hash = BlockFileCache::hash(path.filename().native());
115
75.0k
        BlockFileCache* file_cache_ptr = FileCacheFactory::instance()->get_by_path(path_hash);
116
117
75.0k
        bool has_enough_file_cache_space = opts->allow_adaptive_file_cache_write &&
118
75.1k
                                           config::enable_file_cache_adaptive_write &&
119
75.1k
                                           (opts->approximate_bytes_to_write > 0) &&
120
75.0k
                                           (file_cache_ptr->approximate_available_cache_size() >
121
6.45k
                                            opts->approximate_bytes_to_write);
122
123
18.4E
        VLOG_DEBUG << "path:" << path.filename().native()
124
18.4E
                   << ", write_file_cache:" << opts->write_file_cache
125
18.4E
                   << ", allow_adaptive_file_cache_write:" << opts->allow_adaptive_file_cache_write
126
18.4E
                   << ", has_enough_file_cache_space:" << has_enough_file_cache_space
127
18.4E
                   << ", approximate_bytes_to_write:" << opts->approximate_bytes_to_write
128
18.4E
                   << ", file_cache_available_size:"
129
18.4E
                   << file_cache_ptr->approximate_available_cache_size();
130
75.0k
        if (opts->write_file_cache || has_enough_file_cache_space) {
131
63.7k
            _cache_builder = std::make_unique<FileCacheAllocatorBuilder>(FileCacheAllocatorBuilder {
132
18.4E
                    opts ? opts->is_cold_data : false, opts ? opts->file_cache_expiration_time : 0,
133
63.7k
                    path_hash, file_cache_ptr});
134
63.7k
        }
135
75.0k
        return;
136
76.8k
    }
137
138
    std::unique_ptr<FileCacheAllocatorBuilder> _cache_builder =
139
            nullptr; // nullptr if disable write file cache
140
};
141
142
} // namespace doris::io