Coverage Report

Created: 2026-08-27 11:45

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/fs/packed_file_manager.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 <butil/macros.h>
21
#include <gen_cpp/cloud.pb.h>
22
#include <glog/logging.h>
23
24
#include <atomic>
25
#include <chrono>
26
#include <condition_variable>
27
#include <map>
28
#include <memory>
29
#include <mutex>
30
#include <optional>
31
#include <string>
32
#include <thread>
33
#include <unordered_map>
34
#include <vector>
35
36
#include "common/status.h"
37
#include "io/fs/file_system.h"
38
#include "io/fs/file_writer.h"
39
#include "io/fs/path.h"
40
#include "util/slice.h"
41
42
namespace doris::io {
43
44
struct PackedSliceLocation {
45
    std::string packed_file_path;
46
    int64_t offset;
47
    int64_t size;
48
    int64_t create_time = 0;
49
    int64_t tablet_id = 0;
50
    std::string rowset_id;
51
    std::string resource_id;
52
    int64_t txn_id = 0;
53
    int64_t packed_file_size = -1; // Total size of the packed file, -1 means not set
54
};
55
56
struct PackedAppendContext {
57
    std::string resource_id;
58
    int64_t tablet_id = 0;
59
    std::string rowset_id;
60
    int64_t first_segment_id = 0;
61
    int64_t txn_id = 0;
62
    uint64_t expiration_time = 0; // TTL expiration time in seconds since epoch, 0 means no TTL
63
    bool write_file_cache = true; // Whether to write data to file cache
64
};
65
66
// Global object that manages packing small files into larger files for S3 optimization
67
class PackedFileManager {
68
    struct PackedFileContext;
69
70
public:
71
    static PackedFileManager* instance();
72
73
    // Initialize manager state; file system will be resolved lazily
74
    Status init();
75
76
    // Write a small file to the current packed file
77
    Status append_small_file(const std::string& path, const Slice& data,
78
                             const PackedAppendContext& info);
79
80
    // Block until the small file's packed file is uploaded to S3
81
    Status wait_upload_done(const std::string& path);
82
83
    // Get packed file index information for a small file
84
    Status get_packed_slice_location(const std::string& path, PackedSliceLocation* location);
85
86
    // Start the background management thread
87
    void start_background_manager();
88
89
    // Stop the background management thread
90
    void stop_background_manager();
91
92
    // Mark current packed file for upload and create new one
93
    Status mark_current_packed_file_for_upload(const std::string& resource_id);
94
95
    // Internal helper; expects caller holds _current_packed_file_mutex
96
    Status mark_current_packed_file_for_upload_locked(const std::string& resource_id);
97
98
    void record_packed_file_metrics(const PackedFileContext& packed_file);
99
100
private:
101
40
    PackedFileManager() = default;
102
    ~PackedFileManager();
103
104
    DISALLOW_COPY_AND_ASSIGN(PackedFileManager);
105
106
    // Background thread function for managing packed file lifecycle
107
    void background_manager();
108
109
    // Upload packed file to S3 and update meta service
110
    Status finalize_packed_file_upload(const std::string& packed_file_path, FileWriter* writer);
111
112
    // Update meta service with packed file information
113
    // table_id is used for rate limiting; -1 means no specific table (cross-table operation)
114
    Status update_meta_service(const std::string& packed_file_path,
115
                               const cloud::PackedFileInfoPB& packed_file_info,
116
                               int64_t table_id = -1);
117
118
    // Process uploading files
119
    void process_uploading_packed_files();
120
121
    // Clean up expired data
122
    void cleanup_expired_data();
123
124
    // Internal structure to track packed file state
125
    enum class PackedFileState {
126
        INIT,            // Initial state, no files written yet
127
        ACTIVE,          // Has files but doesn't meet upload conditions
128
        READY_TO_UPLOAD, // Ready for upload, metadata still being prepared
129
        UPLOADING,       // Upload triggered, waiting for writer close to finish
130
        UPLOADED,        // Upload completed
131
        FAILED           // Upload failed
132
    };
133
134
    struct PackedFileContext {
135
        std::string packed_file_path;
136
        std::unique_ptr<FileWriter> writer;
137
        std::unordered_map<std::string, PackedSliceLocation> slice_locations;
138
        int64_t current_offset = 0;
139
        int64_t total_size = 0;
140
        int64_t create_time;
141
        int64_t upload_time = 0;
142
        std::chrono::steady_clock::time_point create_timestamp;
143
        std::optional<std::chrono::steady_clock::time_point> first_append_timestamp;
144
        std::optional<std::chrono::steady_clock::time_point> ready_to_upload_timestamp;
145
        std::optional<std::chrono::steady_clock::time_point> uploading_timestamp;
146
        std::atomic<PackedFileState> state {PackedFileState::INIT};
147
        std::condition_variable upload_cv;
148
        std::mutex upload_mutex;
149
        std::string last_error;
150
        std::string resource_id;
151
        FileSystemSPtr file_system;
152
    };
153
154
    // Create a new packed file state with file writer
155
    Status create_new_packed_file_context(const std::string& resource_id,
156
                                          std::unique_ptr<PackedFileContext>& packed_file_ctx);
157
158
    Status ensure_file_system(const std::string& resource_id, FileSystemSPtr* file_system);
159
160
    // Helper function to wait for packed file upload completion
161
    Status wait_for_packed_file_upload(PackedFileContext* packed_file_ptr);
162
163
    // Thread management
164
    std::atomic<bool> _stop_background_thread {false};
165
    std::unique_ptr<std::thread> _background_thread;
166
167
    // File system
168
    FileSystemSPtr _default_file_system;
169
    std::unordered_map<std::string, FileSystemSPtr> _file_systems;
170
    std::mutex _file_system_mutex;
171
172
    // Current active packed file
173
    std::unordered_map<std::string, std::unique_ptr<PackedFileContext>> _current_packed_files;
174
    std::timed_mutex _current_packed_file_mutex;
175
176
    // Merge files ready for upload or being processed
177
    std::unordered_map<std::string, std::shared_ptr<PackedFileContext>> _uploading_packed_files;
178
179
    // Uploaded packed files (kept for some time for wait_write_done)
180
    std::unordered_map<std::string, std::shared_ptr<PackedFileContext>> _uploaded_packed_files;
181
    std::mutex _packed_files_mutex;
182
183
    // Global index mapping small file path to packed file index
184
    std::unordered_map<std::string, PackedSliceLocation> _global_slice_locations;
185
    std::mutex _global_index_mutex;
186
187
#ifdef BE_TEST
188
public:
189
    void reset_packed_file_bvars_for_test() const;
190
    int64_t packed_file_total_count_for_test() const;
191
    int64_t packed_file_total_small_file_num_for_test() const;
192
    int64_t packed_file_total_size_bytes_for_test() const;
193
    double packed_file_avg_small_file_num_for_test() const;
194
    double packed_file_avg_file_size_for_test() const;
195
    void record_packed_file_metrics_for_test(const PackedFileContext* packed_file);
196
197
    // Test-only helpers to introspect/clear internal state
198
    void clear_state_for_test();
199
    auto& current_packed_files_for_test() { return _current_packed_files; }
200
    auto& uploading_packed_files_for_test() { return _uploading_packed_files; }
201
    auto& uploaded_packed_files_for_test() { return _uploaded_packed_files; }
202
    auto& global_slice_locations_for_test() { return _global_slice_locations; }
203
    auto& file_systems_for_test() { return _file_systems; }
204
    FileSystemSPtr& default_file_system_for_test() { return _default_file_system; }
205
    Status create_new_packed_file_state_for_test(const std::string& resource_id,
206
                                                 std::unique_ptr<PackedFileContext>& ctx) {
207
        return create_new_packed_file_context(resource_id, ctx);
208
    }
209
#endif
210
};
211
212
} // namespace doris::io