github-actions[bot] commented on code in PR #66350: URL: https://github.com/apache/doris/pull/66350#discussion_r3714791852
########## common/cpp/client/obj_storage_client.cpp: ########## @@ -0,0 +1,298 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "obj_storage_client.h" + +#include <cpp/sync_point.h> + +#include <algorithm> +#include <iterator> + +namespace doris { +namespace { + +ObjStorageRateLimitToken acquire_rate_limit( + const std::shared_ptr<const ObjStorageRateLimitPolicy>& policy, ObjStorageRequestType type, + size_t estimated_bytes = 0) { + if (!policy) { + return {}; + } + return policy->acquire(type, estimated_bytes); +} + +} // namespace + +ObjStorageRateLimitToken ObjStorageClient::acquire(ObjStorageRequestType type, + size_t estimated_bytes) const { + return acquire_rate_limit(rate_limit_policy_, type, estimated_bytes); +} + +ObjectStorageUploadResponse ObjStorageClient::create_multipart_upload( + const ObjectStoragePathOptions& opts) { + auto rate_limit = acquire(ObjStorageRequestType::PUT); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return backend_->create_multipart_upload(opts); +} + +ObjectStorageResponse ObjStorageClient::put_object(const ObjectStoragePathOptions& opts, + std::string_view stream) { + auto rate_limit = acquire(ObjStorageRequestType::PUT, stream.size()); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + return backend_->put_object(opts, stream); +} + +ObjectStorageUploadResponse ObjStorageClient::upload_part(const ObjectStoragePathOptions& opts, + std::string_view stream, int part_num) { + auto rate_limit = acquire(ObjStorageRequestType::PUT, stream.size()); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return backend_->upload_part(opts, stream, part_num); +} + +ObjectStorageResponse ObjStorageClient::complete_multipart_upload( + const ObjectStoragePathOptions& opts, + const std::vector<ObjectCompleteMultiPart>& completed_parts) { + auto rate_limit = acquire(ObjStorageRequestType::PUT); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + return backend_->complete_multipart_upload(opts, completed_parts); +} + +ObjectStorageHeadResponse ObjStorageClient::head_object(const ObjectStoragePathOptions& opts) { + auto rate_limit = acquire(ObjStorageRequestType::GET); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return backend_->head_object(opts); +} + +ObjectStorageResponse ObjStorageClient::get_object(const ObjectStoragePathOptions& opts, + void* buffer, size_t offset, size_t bytes_read, + size_t* size_return) { + auto rate_limit = acquire(ObjStorageRequestType::GET, bytes_read); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + auto response = backend_->get_object(opts, buffer, offset, bytes_read, size_return); + if (response.ok()) { + rate_limit.settle_bytes(*size_return); + } + return response; +} + +ObjectStorageListPage ObjStorageClient::list_objects(const ObjectStoragePathOptions& opts, + std::string_view continuation_token) { + auto rate_limit = acquire(ObjStorageRequestType::GET); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return backend_->list_objects(opts, continuation_token); +} + +ObjectStorageResponse ObjectListIterator::has_next() { + if (!is_valid_) { + return { + .status = {TStatusCode::INTERNAL_ERROR, "Iterator is invalid"}, + .http_code = 0, + }; + } + while (next_index_ == objects_.size()) { + if (!has_more_) { + return { + .status = {TStatusCode::END_OF_FILE, "No more results"}, + .http_code = 200, + }; + } + auto page = client_->list_objects(opts_, continuation_token_); + if (!page.resp.ok()) { + is_valid_ = false; + return page.resp; + } + objects_ = std::move(page.objects); + next_index_ = 0; + continuation_token_ = std::move(page.continuation_token); + has_more_ = page.has_more; + } + return ObjectStorageResponse::OK(); +} + +ObjectStorageListResponse ObjectListIterator::next() { + auto response = has_next(); + if (response.status.code == ObjectStorageStatus::END_OF_FILE) { + return {.resp = ObjectStorageResponse::OK(), .results_ = {}}; + } + if (!response.ok()) { + return {.resp = std::move(response), .results_ = {}}; + } + return { + .resp = ObjectStorageResponse::OK(), + .results_ = std::move(objects_[next_index_++]), + }; +} + +ObjectStorageResponse ObjStorageClient::delete_objects(const ObjectStoragePathOptions& opts, + std::vector<std::string> objs) { + const auto max_batch_size = std::max<size_t>(1, backend_->capabilities().max_delete_batch); + for (size_t begin = 0; begin < objs.size(); begin += max_batch_size) { + const auto end = std::min(begin + max_batch_size, objs.size()); + auto rate_limit = acquire(ObjStorageRequestType::PUT); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + std::vector<std::string> batch(std::make_move_iterator(objs.begin() + begin), + std::make_move_iterator(objs.begin() + end)); + auto response = backend_->delete_objects(opts, std::move(batch)); + if (!response.ok()) { + return response; + } + } + return ObjectStorageResponse::OK(); +} + +ObjectStorageResponse ObjStorageClient::delete_object(const ObjectStoragePathOptions& opts) { + auto rate_limit = acquire(ObjStorageRequestType::PUT); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + return backend_->delete_object(opts); +} + +ObjectStorageResponse ObjStorageClient::delete_objects_recursively( + const ObjectStoragePathOptions& opts, const RecursiveDeleteOptions& options) { + auto list_opts = opts; + if (list_opts.prefix.empty()) { + list_opts.prefix = list_opts.key; + } + auto delete_batch_size = std::max<size_t>(1, backend_->capabilities().max_delete_batch); + TEST_SYNC_POINT_CALLBACK("ObjStorageClient::delete_objects_recursively_", &delete_batch_size); + delete_batch_size = std::max<size_t>(1, delete_batch_size); + const auto max_tasks_per_batch = std::max<size_t>(1, options.max_tasks_per_batch); + std::vector<std::string> keys; + keys.reserve(delete_batch_size); + std::vector<ObjStorageDeleteTask> tasks; + tasks.reserve(max_tasks_per_batch); + + auto add_delete_task = [&]() { + tasks.emplace_back([backend = backend_, rate_limit_policy = rate_limit_policy_, Review Comment: [P2] Keep this recursive delete streaming into the bounded executor. Production uses `max_tasks_per_batch=1000`, and each S3 task captures up to 1000 keys, so the new code retains as many as one million key strings and performs up to 1000 list requests before any delete starts. The removed path submitted each full batch immediately; the 40-thread pool's bounded queue supplied backpressure while workers freed key vectors and list and delete work overlapped. Please preserve that bounded pipelining (or introduce a much smaller independent key-memory bound) while keeping the configured synchronization boundary. ########## cloud/test/s3_accessor_mock_test.cpp: ########## @@ -66,21 +66,21 @@ class MockS3Client : public Aws::S3::S3Client { }; TEST_F(S3AccessorMockTest, list_objects_compatibility) { - // If storage only supports ListObjectsV1, s3_obj_storage_client.list_objects + // If storage only supports ListObjectsV1, s3_obj_storage_backend.list_objects // should return an error. auto mock_s3_client = std::make_shared<MockS3Client>(); - S3ObjClient s3_obj_client(mock_s3_client, "dummy-endpoint"); + S3ObjStorageBackend s3_obj_client(mock_s3_client, {.endpoint = "dummy-endpoint"}); ListObjectsV2Result result; result.SetIsTruncated(true); EXPECT_CALL(*mock_s3_client, ListObjectsV2(testing::_)) .WillOnce(testing::Return(ListObjectsV2Outcome(result))); - auto response = s3_obj_client.list_objects( - {.bucket = "dummy-bucket", .key = "S3AccessorMockTest/list_objects_compatibility"}); + auto page = s3_obj_client.list_objects( + {.bucket = "dummy-bucket", .key = "S3AccessorMockTest/list_objects_compatibility"}, {}); - EXPECT_FALSE(response->has_next()); - EXPECT_FALSE(response->is_valid()); + EXPECT_NE(page.resp.status.code, TStatusCode::OK); Review Comment: [P1] Keep this assertion linkable in the Cloud UT target. `status.code` is an `int`, but the RHS is `TStatusCode::type`, so gtest instantiates the enum stream printer; live Cloud UT build 1015520 currently fails this target with an undefined reference to `doris::operator<<(std::ostream&, doris::TStatusCode::type const&)`. Please compare through `page.resp.ok()` or cast the enum to the stored integer type so this test binary can link. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
