plusplusjiajia commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4237137194
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -338,30 +364,38 @@ Status ArrowS3FileIO::DeleteFile(const std::string&
file_location) {
}
Status ArrowS3FileIO::DeleteFiles(const std::vector<std::string>&
file_locations) {
- // One snapshot so the whole batch matches the same delegate generation; only
- // ever a handful of delegates, so a linear scan beats hashing.
+ // Like Java's S3FileIO: delete on the FileIO's bounded pool, keep going
after
+ // a failure and report the count. Arrow has no batch delete for S3. One
+ // snapshot so the whole batch matches the same delegate generation.
std::shared_ptr<ArrowFileSystemFileIO> fallback;
DelegatesByPrefix by_prefix;
{
std::shared_lock lock(mutex_);
fallback = default_file_io_;
by_prefix = file_io_by_prefix_;
}
- std::vector<std::pair<std::shared_ptr<ArrowFileSystemFileIO>,
std::vector<std::string>>>
- locations_by_io;
+ // Each failure is logged where the caller logs and only counted, so the
+ // returned error stays one line however large the batch.
+ auto logger = GetCurrentLogger();
+ std::atomic<size_t> failed = 0;
+ TaskGroup group;
+ group.SetExecutor(std::ref(delete_executor_));
for (const auto& file_location : file_locations) {
- auto file_io = MatchDelegate(fallback, by_prefix, file_location);
- auto it = std::ranges::find_if(
- locations_by_io, [&](const auto& entry) { return entry.first ==
file_io; });
- if (it == locations_by_io.end()) {
- locations_by_io.emplace_back(std::move(file_io),
- std::vector<std::string>{file_location});
- } else {
- it->second.push_back(file_location);
- }
+ group.Submit([file_io = MatchDelegate(fallback, by_prefix, file_location),
Review Comment:
Thanks! Deletes now use at most 64 workers sharing the file list, so task
and future counts stay bounded as the number of files grows. Added a regression
test.
##########
mkdocs/docs/file-io.md:
##########
@@ -66,6 +66,7 @@ each file location's scheme.
| `client.region` | `us-east-1` | Region to sign requests for |
| `s3.endpoint` | `https://127.0.0.1:9000` | Endpoint to use instead of the
AWS one. When absent, the `AWS_ENDPOINT_URL_S3` / `AWS_ENDPOINT_URL`
environment variables are consulted |
| `s3.path-style-access` | `true` | Address buckets as a path
(`endpoint/bucket`) instead of a virtual host (`bucket.endpoint`). Only takes
effect together with a custom endpoint |
+| `s3.delete.num-threads` | `8` | Size of the thread pool `DeleteFiles` runs
on, shared by every call on the FileIO. Defaults to the number of hardware
threads |
Review Comment:
Removed the S3 property and documented the executor configuration API
instead.
--
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]
