github-actions[bot] commented on code in PR #66350: URL: https://github.com/apache/doris/pull/66350#discussion_r3773623706
########## common/cpp/obj-client/rate_limited_obj_storage_client.cpp: ########## @@ -0,0 +1,156 @@ +// 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 "rate_limited_obj_storage_client.h" + +namespace doris { + +ObjStorageAdmission RateLimitedObjStorageClient::acquire(S3RateLimitType type, + size_t estimated_bytes) const { + return rate_limit_policy_->acquire(type, estimated_bytes); +} + +ObjStorageUploadResult RateLimitedObjStorageClient::create_multipart_upload( + const ObjStoragePath& opts) { + auto rate_limit = acquire(S3RateLimitType::PUT); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return inner_->create_multipart_upload(opts); +} + +ObjStorageResponse RateLimitedObjStorageClient::put_object(const ObjStoragePath& opts, + std::string_view stream) { + auto rate_limit = acquire(S3RateLimitType::PUT, stream.size()); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + return inner_->put_object(opts, stream); +} + +ObjStorageUploadResult RateLimitedObjStorageClient::upload_part(const ObjStoragePath& opts, + const std::string& upload_id, + std::string_view stream, + int part_num) { + auto rate_limit = acquire(S3RateLimitType::PUT, stream.size()); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return inner_->upload_part(opts, upload_id, stream, part_num); +} + +ObjStorageResponse RateLimitedObjStorageClient::complete_multipart_upload( + const ObjStoragePath& opts, const std::string& upload_id, + const std::vector<ObjStorageCompletedPart>& completed_parts) { + auto rate_limit = acquire(S3RateLimitType::PUT); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + return inner_->complete_multipart_upload(opts, upload_id, completed_parts); +} + +ObjStorageHeadResult RateLimitedObjStorageClient::head_object(const ObjStoragePath& opts) { + auto rate_limit = acquire(S3RateLimitType::GET); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return inner_->head_object(opts); +} + +ObjStorageResponse RateLimitedObjStorageClient::get_object(const ObjStoragePath& opts, void* buffer, + size_t offset, size_t bytes_read, + size_t* size_return) { + auto rate_limit = acquire(S3RateLimitType::GET, bytes_read); + if (!rate_limit.resp.ok()) { + return rate_limit.resp; + } + auto response = inner_->get_object(opts, buffer, offset, bytes_read, size_return); + if (response.ok()) { + rate_limit.settle_bytes(*size_return); + } + return response; +} + +ObjStorageListPageResult RateLimitedObjStorageClient::list_objects_page( + const ObjStoragePath& opts, std::string_view continuation_token) { + auto rate_limit = acquire(S3RateLimitType::GET); + if (!rate_limit.resp.ok()) { + return {.resp = std::move(rate_limit.resp)}; + } + return inner_->list_objects_page(opts, continuation_token); +} + +ObjStorageResponse RateLimitedObjStorageClient::delete_objects(const ObjStoragePath& opts, + std::vector<std::string> objs) { + auto rate_limit = acquire(S3RateLimitType::PUT); Review Comment: [P2] Admit every provider batch in direct bulk deletes This acquires one PUT token for the whole vector, but the S3 provider below splits it into 1,000-key `DeleteObjects` calls and Azure splits it into 256-key `SubmitBatch` calls. `S3Accessor::delete_files` passes unbounded rowset/resource vectors here, and the deleted Recycler clients acquired inside each provider batch, so (for example) 2,001 S3 keys now produce three SDK requests after only one request-level admission. This is separate from the recursive-delete thread because it affects direct bulk deletion. Please place batching above the policy-bearing facade or otherwise acquire once per SDK batch, and cover an input exceeding each provider's limit. ########## common/cpp/obj-client/obj_storage_client.cpp: ########## @@ -0,0 +1,205 @@ +// 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 <glog/logging.h> + +#include <algorithm> +#include <chrono> + +namespace doris { +std::unique_ptr<ObjStorageListIterator> ObjStorageClient::list_objects(const ObjStoragePath& opts) { + return std::make_unique<ObjStorageListIterator>(shared_from_this(), opts); +} + +ObjStorageResponse ObjStorageClient::list_objects(const ObjStoragePath& opts, + std::vector<ObjectMeta>* objects) { + objects->clear(); + auto iter = list_objects(opts); + for (;;) { + auto result = iter->next(); + if (!result.object.has_value()) { + if (!result.resp.ok()) { + objects->clear(); + } + return result.resp; + } + objects->emplace_back(std::move(*result.object)); + } +} + +ObjStorageResponse ObjStorageListIterator::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_page(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 ObjStorageResponse::OK(); +} + +ObjStorageListResult ObjStorageListIterator::next() { + auto response = has_next(); + if (response.status.code == ObjStorageStatus::END_OF_FILE) { + return {.resp = ObjStorageResponse::OK(), .object = {}}; + } + if (!response.ok()) { + return {.resp = std::move(response), .object = {}}; + } + return { + .resp = ObjStorageResponse::OK(), + .object = std::move(objects_[next_index_++]), + }; +} + +ObjStorageResponse ObjStorageClient::delete_objects_recursively( + const ObjStoragePath& opts, const ObjStorageRecursiveDeleteOptions& options) { + return delete_objects_recursively_impl(opts, options); +} + +ObjStorageResponse ObjStorageClient::delete_objects_recursively_impl( + const ObjStoragePath& opts, const ObjStorageRecursiveDeleteOptions& options) { + const auto start_time = std::chrono::steady_clock::now(); + auto list_opts = opts; + if (list_opts.prefix.empty()) { + list_opts.prefix = list_opts.key; + } + auto delete_batch_size = std::max<size_t>(1, 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); + size_t pending_tasks = 0; + size_t total_batches = 0; + size_t num_deleted = 0; + size_t error_count = 0; + auto first_error = ObjStorageResponse::OK(); + + auto elapsed_milliseconds = [&]() { + return std::chrono::duration_cast<std::chrono::milliseconds>( + std::chrono::steady_clock::now() - start_time) + .count(); + }; + auto finish = [&](ObjStorageResponse response) { + LOG(INFO) << "delete objects under " << list_opts.bucket << "/" << list_opts.prefix + << " finished, ret=" << response.status.code + << ", total_batches=" << total_batches << ", num_deleted=" << num_deleted + << ", error_count=" << error_count << ", cost=" << elapsed_milliseconds() + << " ms"; + return response; + }; + auto record_error = [&](ObjStorageResponse response) { + if (response.ok()) { + return; + } + ++error_count; + if (first_error.ok()) { + first_error = std::move(response); + } + }; + + auto wait_for_tasks = [&]() { + if (pending_tasks == 0) { + return ObjStorageResponse::OK(); + } + const auto tasks_in_batch = pending_tasks; + pending_tasks = 0; + auto response = options.executor ? options.executor->wait() : ObjStorageResponse::OK(); + ++total_batches; + LOG(INFO) << "delete objects under " << list_opts.bucket << "/" << list_opts.prefix + << " batch " << total_batches << " completed" + << ", tasks_in_batch=" << tasks_in_batch << ", total_deleted=" << num_deleted + << ", elapsed=" << elapsed_milliseconds() << " ms"; + return response; + }; + auto submit_delete_task = [&]() { + ObjStorageDeleteTask task = [this, bucket = opts.bucket, + batch = std::move(keys)]() mutable { + return delete_objects(ObjStoragePath {.bucket = std::move(bucket)}, std::move(batch)); + }; + keys.clear(); + keys.reserve(delete_batch_size); + + ObjStorageResponse response; + if (options.executor) { + response = options.executor->submit(std::move(task)); + } else { + response = task(); + } + ++pending_tasks; + if (!response.ok()) { + record_error(std::move(response)); + record_error(wait_for_tasks()); + return; + } + if (pending_tasks == max_tasks_per_batch) { Review Comment: [P2] Stop recursive deletion after a failed executor batch At this boundary, `wait_for_tasks()` can report that the production Recycler executor cancelled the batch, but `submit_delete_task()` only records that response and the outer loop resumes listing and scheduling later destructive batches. The new `ProductionExecutorContinuesAfterCancellation` test demonstrates this by expecting all six pages and a second delete call after the first batch fails; the removed Recycler loop broke immediately when `when_all` reported `finished == false`. Please propagate the failed wait out of this helper, drain the current batch, and stop before starting later list/delete batches. -- 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]
