plusplusjiajia commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4226927599
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -181,9 +185,11 @@ std::string CanonicalizeS3Scheme(std::string_view
location) {
class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials {
public:
ArrowS3FileIO(std::shared_ptr<::arrow::fs::FileSystem> arrow_fs,
- std::unordered_map<std::string, std::string>
default_properties)
+ std::unordered_map<std::string, std::string>
default_properties,
+ size_t delete_threads)
:
default_file_io_(std::make_shared<ArrowFileSystemFileIO>(std::move(arrow_fs))),
- default_properties_(std::move(default_properties)) {}
+ default_properties_(std::move(default_properties)),
+ delete_threads_(delete_threads) {}
Review Comment:
Thanks! That member was removed; the FileIO now owns the thread pool.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -237,6 +243,7 @@ class ArrowS3FileIO final : public FileIO, public
SupportsStorageCredentials {
std::shared_ptr<ArrowFileSystemFileIO> default_file_io_;
std::unordered_map<std::string, std::string> default_properties_;
+ size_t delete_threads_;
Review Comment:
Thanks! That member was removed; the FileIO now owns the thread pool.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -284,12 +292,37 @@ Status ArrowS3FileIO::DeleteFile(const std::string&
file_location) {
}
Status ArrowS3FileIO::DeleteFiles(const std::vector<std::string>&
file_locations) {
- std::unordered_map<ArrowFileSystemFileIO*, std::vector<std::string>>
locations_by_io;
- for (const auto& file_location : file_locations) {
- locations_by_io[&FileIOForPath(file_location)].push_back(file_location);
+ // Like Java's S3FileIO: delete concurrently, keep going after a failure and
+ // report the count. Arrow has no batch delete for S3.
+ std::atomic<size_t> next = 0;
+ std::atomic<size_t> failed = 0;
+ auto delete_remaining = [&] {
+ for (size_t i = next++; i < file_locations.size(); i = next++) {
+ const auto& file_location = file_locations[i];
+ if (auto status = FileIOForPath(file_location).DeleteFile(file_location);
+ !status.has_value()) {
+ ICEBERG_LOG_WARN("Failed to delete {}: {}", file_location,
+ status.error().message);
+ ++failed;
+ }
+ }
+ };
+ // Plus the calling thread. Helpers log where the caller does.
+ auto logger = GetCurrentLogger();
+ std::vector<std::future<void>> helpers;
+ for (size_t i = 1; i < std::min(delete_threads_, file_locations.size());
++i) {
+ helpers.push_back(std::async(std::launch::async, [&] {
Review Comment:
@kamcheungting-db Thanks! The FileIO now owns a bounded Arrow thread pool
shared across DeleteFiles calls, with tasks run through TaskGroup.
--
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]
