Coverage Report

Created: 2026-08-18 12:47

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
common/cpp/obj-client/obj_storage_client.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 <gen_cpp/Status_types.h>
21
22
#include <cstdint>
23
#include <filesystem>
24
#include <functional>
25
#include <memory>
26
#include <optional>
27
#include <string>
28
#include <string_view>
29
#include <utility>
30
#include <vector>
31
32
namespace doris {
33
// Names are in lexico order.
34
enum class ObjStorageProvider : uint8_t {
35
    UNKNOWN = 0,
36
    AWS = 1,
37
    AZURE = 2,
38
    BOS = 3,
39
    COS = 4,
40
    OSS = 5,
41
    OBS = 6,
42
    GCP = 7,
43
    TOS = 8,
44
};
45
46
/// eg:
47
///     s3://bucket1/path/to/file.txt
48
/// path:   s3://bucket1/path/to/file.txt
49
/// bucket: bucket1
50
/// key:    path/to/file.txt
51
struct ObjStoragePath {
52
    std::filesystem::path path = "";
53
    std::string bucket {}; // blob container in azure
54
    std::string key {};    // blob name in azure
55
    std::string prefix {}; // for list and recursive delete
56
};
57
58
struct ObjStorageEndpointInfo {
59
    std::string endpoint {};
60
    std::string ak {};
61
    std::string sk {};
62
    std::string tls_debug_context {};
63
};
64
65
struct ObjectMeta {
66
    std::string key {};
67
    int64_t size {0};
68
    int64_t mtime_s {0};
69
};
70
71
struct ObjStorageCompletedPart {
72
    int part_num = 0;
73
    std::string etag {};
74
};
75
76
struct ObjStorageStatus {
77
    enum Code : int {
78
        OK = TStatusCode::OK,
79
        INTERNAL_ERROR = TStatusCode::INTERNAL_ERROR,
80
        LIMIT_REACH = TStatusCode::LIMIT_REACH,
81
        NOT_FOUND = TStatusCode::NOT_FOUND,
82
        END_OF_FILE = TStatusCode::END_OF_FILE,
83
        IO_ERROR = TStatusCode::IO_ERROR,
84
        NETWORK_ERROR = TStatusCode::NETWORK_ERROR,
85
        // Keep the legacy BE error code for access-denied object storage responses.
86
        PERMISSION_DENIED = -256,
87
    };
88
89
1.21k
    ObjStorageStatus(int r = OK, std::string msg = "") : code(r), msg(std::move(msg)) {}
90
    // clang-format off
91
    int code {OK}; // To unify the error handle logic with BE, we'd better use the same error code as BE
92
    // clang-format on
93
    std::string msg;
94
};
95
96
// We only store error code along with err_msg instead of Status to unify BE and recycler's error handle logic
97
struct ObjStorageResponse {
98
    ObjStorageStatus status {ObjStorageStatus::OK, ""};
99
    int http_code {200};
100
    std::string request_id {};
101
0
    static ObjStorageResponse OK() {
102
        // clang-format off
103
0
        return {
104
0
                .status = ObjStorageStatus{ObjStorageStatus::OK, ""},
105
0
                .http_code = 200,
106
0
                .request_id = ""
107
0
        };
108
        // clang-format on
109
0
    }
110
111
0
    static ObjStorageResponse rate_limit(int status_code, int http_code, std::string message) {
112
0
        return {
113
0
                .status = ObjStorageStatus {status_code, std::move(message)},
114
0
                .http_code = http_code,
115
0
        };
116
0
    }
117
118
365k
    bool ok() const { return status.code == ObjStorageStatus::OK; }
119
};
120
121
// Convert a provider HTTP response code into the object-storage status domain. A non-positive
122
// response code means that no HTTP response was received.
123
ObjStorageStatus obj_storage_status_from_http_code(int http_code, std::string message);
124
125
struct ObjStorageUploadResult {
126
    ObjStorageResponse resp = ObjStorageResponse::OK();
127
    std::optional<std::string> upload_id = std::nullopt;
128
    std::optional<std::string> etag = std::nullopt;
129
};
130
131
struct ObjStorageHeadResult {
132
    ObjStorageResponse resp = ObjStorageResponse::OK();
133
    long long file_size {0};
134
};
135
136
struct ObjStorageListResult {
137
    ObjStorageResponse resp = ObjStorageResponse::OK();
138
    std::optional<ObjectMeta> object = std::nullopt;
139
};
140
141
struct ObjStorageListPageResult {
142
    ObjStorageResponse resp = ObjStorageResponse::OK();
143
    std::vector<ObjectMeta> objects {};
144
    std::string continuation_token {};
145
    bool has_more = false;
146
};
147
148
struct ObjStorageCapabilities {
149
    size_t max_delete_batch = 1;
150
    size_t max_list_page = 1000;
151
};
152
153
using ObjStorageDeleteTask = std::function<ObjStorageResponse()>;
154
155
// Implementations must enqueue tasks immediately and remain reusable after wait() completes.
156
// Recycler implements this in cloud/src/recycler/s3_accessor.cpp; BE uses the synchronous fallback.
157
class ObjStorageDeleteExecutor {
158
public:
159
    virtual ~ObjStorageDeleteExecutor() = default;
160
    virtual ObjStorageResponse submit(ObjStorageDeleteTask task) = 0;
161
    virtual ObjStorageResponse wait() = 0;
162
};
163
164
struct ObjStorageRecursiveDeleteOptions {
165
    int64_t expiration_time = 0;
166
    size_t max_tasks_per_batch = 1;
167
    std::shared_ptr<ObjStorageDeleteExecutor> executor {};
168
};
169
170
class ObjStorageListIterator;
171
class RateLimitedObjStorageClient;
172
173
// Provider implementations are in common/cpp/obj-client/s3_obj_storage_client.cpp and
174
// common/cpp/obj-client/azure_obj_storage_client.cpp.
175
// Clients are shared-owned so a lazy list iterator can keep the complete decorator chain alive.
176
class ObjStorageClient : public std::enable_shared_from_this<ObjStorageClient> {
177
public:
178
    ObjStorageClient() = default;
179
0
    virtual ~ObjStorageClient() = default;
180
181
    ObjStorageClient(const ObjStorageClient&) = delete;
182
    ObjStorageClient& operator=(const ObjStorageClient&) = delete;
183
    // Create a multipart upload request. The returned token may be provider-issued or local and
184
    // identifies this writer's parts.
185
    // The input parameters should include the bucket and key for the object storage.
186
    virtual ObjStorageUploadResult create_multipart_upload(const ObjStoragePath& opts) = 0;
187
    // To directly upload a piece of data to object storage and generate a user-visible file.
188
    // You need to clearly specify the bucket and key
189
    virtual ObjStorageResponse put_object(const ObjStoragePath& opts, std::string_view stream) = 0;
190
    // Upload one part of a large object without making the object visible to users. upload_id is
191
    // the provider-issued or local writer token returned by create_multipart_upload.
192
    // Reusing the same bucket, key, upload_id, and part_num replaces that writer's staged part.
193
    virtual ObjStorageUploadResult upload_part(const ObjStoragePath& opts,
194
                                               const std::string& upload_id,
195
                                               std::string_view stream, int part_num) = 0;
196
    // Combine the parts belonging to upload_id into the key passed in opts. After this succeeds,
197
    // the complete object is visible in object storage.
198
    virtual ObjStorageResponse complete_multipart_upload(
199
            const ObjStoragePath& opts, const std::string& upload_id,
200
            const std::vector<ObjStorageCompletedPart>& completed_parts) = 0;
201
    // According to the passed bucket and key, it will access whether the corresponding file exists in the object storage.
202
    // If it exists, it will return the corresponding file size
203
    virtual ObjStorageHeadResult head_object(const ObjStoragePath& opts) = 0;
204
    // According to the bucket and key, it finds the corresponding file in the object storage
205
    // and starting from the offset, it reads bytes_read bytes into the buffer, with size_return recording the actual number of bytes read
206
    virtual ObjStorageResponse get_object(const ObjStoragePath& opts, void* buffer, size_t offset,
207
                                          size_t bytes_read, size_t* size_return) = 0;
208
    // Return a lazy iterator that fetches one provider page at a time.
209
    virtual std::unique_ptr<ObjStorageListIterator> list_objects(const ObjStoragePath& opts);
210
    // Collect all objects by consuming the lazy iterator. This preserves the eager BE API.
211
    // **Notice**: The files returned by this function contain the full key in object storage.
212
    virtual ObjStorageResponse list_objects(const ObjStoragePath& opts,
213
                                            std::vector<ObjectMeta>* objects);
214
215
    // According to the bucket and prefix specified by the user, it performs batch deletion based on the object names in the object array.
216
    virtual ObjStorageResponse delete_objects(const ObjStoragePath& opts,
217
                                              std::vector<std::string> objs) = 0;
218
    // Delete the file named key in the object storage bucket.
219
    virtual ObjStorageResponse delete_object(const ObjStoragePath& opts) = 0;
220
    virtual ObjStorageCapabilities capabilities() const = 0;
221
    // Return a presigned URL for users to access the object
222
    virtual std::string generate_presigned_url(const ObjStoragePath& opts,
223
                                               int64_t expiration_secs) = 0;
224
225
    // Get the objects' expiration time on the bucket
226
    virtual ObjStorageResponse get_lifecycle(const std::string& bucket,
227
                                             int64_t* expiration_days) = 0;
228
229
    // Check if the objects' versioning is on or off
230
    // returns 0 when versioning is on, otherwise versioning is off or check failed
231
    virtual ObjStorageResponse check_versioning(const std::string& bucket) = 0;
232
233
    virtual ObjStorageResponse abort_multipart_upload(const ObjStoragePath& path,
234
                                                      const std::string& upload_id) = 0;
235
236
protected:
237
    // Fetch at most one page. One call corresponds to exactly one provider request.
238
    virtual ObjStorageListPageResult list_objects_page(const ObjStoragePath& opts,
239
                                                       std::string_view continuation_token) = 0;
240
241
private:
242
    friend class ObjStorageListIterator;
243
    friend class RateLimitedObjStorageClient;
244
};
245
246
// A client-side iterator returned by ObjStorageClient. It requests one fixed-size page at a time.
247
class ObjStorageListIterator {
248
public:
249
    ObjStorageListIterator(std::shared_ptr<ObjStorageClient> client, ObjStoragePath opts)
250
16
            : client_(std::move(client)), opts_(std::move(opts)) {}
251
252
0
    bool is_valid() const { return is_valid_; }
253
    ObjStorageResponse has_next();
254
    ObjStorageListResult next();
255
256
private:
257
    std::shared_ptr<ObjStorageClient> client_;
258
    ObjStoragePath opts_;
259
    std::vector<ObjectMeta> objects_;
260
    size_t next_index_ = 0;
261
    std::string continuation_token_;
262
    bool has_more_ = true;
263
    bool is_valid_ = true;
264
};
265
266
// Provider-independent recursive deletion shared by concrete clients. Passing the complete client
267
// decorator chain keeps it alive for asynchronous delete tasks and applies policy per list page and
268
// delete batch.
269
ObjStorageResponse delete_objects_recursively(std::shared_ptr<ObjStorageClient> client,
270
                                              const ObjStoragePath& path,
271
                                              const ObjStorageRecursiveDeleteOptions& options = {});
272
} // namespace doris
273
274
// Keep the BE namespace spelling source-compatible while the implementation is
275
// shared with Recycler in `doris`.
276
namespace doris::io {
277
using ::doris::ObjStorageCapabilities;
278
using ::doris::ObjStorageClient;
279
using ::doris::ObjStorageCompletedPart;
280
using ::doris::ObjStorageDeleteExecutor;
281
using ::doris::ObjStorageDeleteTask;
282
using ::doris::ObjStorageEndpointInfo;
283
using ::doris::ObjStorageHeadResult;
284
using ::doris::ObjStorageListIterator;
285
using ::doris::ObjStorageListPageResult;
286
using ::doris::ObjStorageListResult;
287
using ::doris::ObjStoragePath;
288
using ::doris::ObjStorageProvider;
289
using ::doris::ObjStorageRecursiveDeleteOptions;
290
using ::doris::ObjStorageResponse;
291
using ::doris::ObjStorageStatus;
292
using ::doris::ObjStorageUploadResult;
293
using ::doris::ObjectMeta;
294
using ::doris::delete_objects_recursively;
295
} // namespace doris::io