You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "plusplusjiajia (via GitHub)" <gi...@apache.org> on 2026/09/26 06:26:07 UTC
[PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
plusplusjiajia opened a new pull request, #966:
URL: https://github.com/apache/iceberg-cpp/pull/966
Follow-up to https://github.com/apache/iceberg-cpp/pull/898#discussion_r4091927837.
`ArrowS3FileIO::DeleteFiles` returned at the first credential prefix whose delete failed, so files under the remaining prefixes were never attempted. Java's `S3FileIO.deleteFiles` attempts every batch, logs each failed path, and throws `BulkDeletionFailureException` with the count.
This deletes each file through its delegate, logs every failure, and returns `Failed to delete N of M files`. The requests are unchanged: Arrow's S3 `DeleteFiles` also deletes one file at a time. `ResolvingFileIO` stays fail-fast across delegates, as Java's does.
`DeleteFilesAttemptsEveryFile` puts an allowed file between two denied ones; it fails on `main` and passes here. This touches the same function as #898, so whichever lands second needs a small rebase.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "zhjwpku (via GitHub)" <gi...@apache.org>.
zhjwpku commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4113680605
##########
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:
We have `TaskGroup` and `Executor` abstractions, not sure if they fit it here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4237136700
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -372,7 +406,22 @@ Result<std::unique_ptr<FileIO>> MakeS3FileIO(
const std::unordered_map<std::string, std::string>& properties) {
// Uses default credentials if properties are empty.
ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
- return std::make_unique<ArrowS3FileIO>(std::move(fs), properties);
+ // Java defaults to one delete thread per processor.
+ int num_delete_threads =
+ std::max(1, static_cast<int>(std::thread::hardware_concurrency()));
+ if (const auto* value = FindProperty(properties, S3Properties::kDeleteNumThreads);
+ value != nullptr) {
+ ICEBERG_ASSIGN_OR_RAISE(num_delete_threads, StringUtils::ParseNumber<int>(*value));
+ if (num_delete_threads <= 0) {
+ return InvalidArgument(R"("{}" must be positive)", S3Properties::kDeleteNumThreads);
+ }
+ }
+ // The pool starts its workers on the first delete, so a FileIO that never
+ // deletes costs no threads.
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(auto delete_pool,
Review Comment:
Thanks! Added SetS3FileIODeleteExecutor for a caller-owned executor shared across instances, including registry-created FileIOs. The default is sequential, and initialization and lifetime requirements are documented.
##########
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),
+ &file_location, &logger, &failed]() -> Status {
+ ScopedLogger bind(logger);
Review Comment:
Thanks! Delete exceptions are now caught and counted per file. Tests cover both standard and unknown exceptions and verify that the remaining files are still attempted.
##########
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),
+ &file_location, &logger, &failed]() -> Status {
+ ScopedLogger bind(logger);
+ if (auto status = file_io->DeleteFile(file_location); !status.has_value()) {
+ ICEBERG_LOG_WARN("Failed to delete {}: {}", file_location,
+ status.error().message);
+ ++failed;
+ }
+ return {};
+ });
}
- for (auto& [file_io, locations] : locations_by_io) {
- ICEBERG_RETURN_UNEXPECTED(file_io->DeleteFiles(locations));
+ ICEBERG_RETURN_UNEXPECTED(std::move(group).Run());
Review Comment:
Thanks! Accepted work is now drained before returning, including submission failures. Added tests for rejected and throwing submissions after work has been accepted.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac merged PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "kamcheungting-db (via GitHub)" <gi...@apache.org>.
kamcheungting-db commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4179796208
##########
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:
```suggestion
num_delete_threads_(delete_threads) {}
```
##########
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:
```suggestion
size_t num_delete_threads_;
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "kamcheungting-db (via GitHub)" <gi...@apache.org>.
kamcheungting-db commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4179833156
##########
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:
@plusplusjiajia
I get your opinion above.
In this case, could ArrowS3FileIO has its own bounded executor initialized from s3.delete.num-threads and reuse it across DeleteFiles calls?
This enforces the concurrency limit across calls and allow using TaskGroup instead of creating threads per invocation.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "zhjwpku (via GitHub)" <gi...@apache.org>.
zhjwpku commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4113682917
##########
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:
Maybe not since there is no need to retry here. cc @HuaHuaY
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4236747662
##########
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),
+ &file_location, &logger, &failed]() -> Status {
+ ScopedLogger bind(logger);
+ if (auto status = file_io->DeleteFile(file_location); !status.has_value()) {
+ ICEBERG_LOG_WARN("Failed to delete {}: {}", file_location,
+ status.error().message);
+ ++failed;
+ }
+ return {};
+ });
}
- for (auto& [file_io, locations] : locations_by_io) {
- ICEBERG_RETURN_UNEXPECTED(file_io->DeleteFiles(locations));
+ ICEBERG_RETURN_UNEXPECTED(std::move(group).Run());
Review Comment:
Please ensure every accepted task completes before DeleteFiles exits, including when submission throws after earlier tasks were accepted. TaskGroup uses promise-backed futures whose destruction does not wait, so unwinding can leave queued tasks referencing destroyed `logger`, `failed`, or input data. Please add exception-safe draining and a test where submission fails after some work has been accepted.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -372,7 +406,22 @@ Result<std::unique_ptr<FileIO>> MakeS3FileIO(
const std::unordered_map<std::string, std::string>& properties) {
// Uses default credentials if properties are empty.
ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
- return std::make_unique<ArrowS3FileIO>(std::move(fs), properties);
+ // Java defaults to one delete thread per processor.
+ int num_delete_threads =
+ std::max(1, static_cast<int>(std::thread::hardware_concurrency()));
+ if (const auto* value = FindProperty(properties, S3Properties::kDeleteNumThreads);
+ value != nullptr) {
+ ICEBERG_ASSIGN_OR_RAISE(num_delete_threads, StringUtils::ParseNumber<int>(*value));
+ if (num_delete_threads <= 0) {
+ return InvalidArgument(R"("{}" must be positive)", S3Properties::kDeleteNumThreads);
+ }
+ }
+ // The pool starts its workers on the first delete, so a FileIO that never
+ // deletes costs no threads.
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(auto delete_pool,
Review Comment:
I don't think each FileIO should create its own thread pool. Java's executor is static and shared across instances. Could we instead expose a static/public configuration API for a caller-owned `iceberg::Executor*`, defaulting to `nullptr` for sequential deletion? This should also work for registry-created FileIOs. Please document initialization and executor lifetime requirements; the executor, rather than each FileIO, should control concurrency.
##########
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:
As my other comment, this should not be a S3FileIO property.
##########
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:
Could we bound the number of in-flight tasks rather than creating a task and future for every file? Large snapshot expirations can contain millions of paths, and TaskGroup retains all futures until completion. A bounded submission window or chunked tasks would avoid this allocation and queue growth while preserving the delegate snapshot and per-file failure accounting.
##########
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),
+ &file_location, &logger, &failed]() -> Status {
+ ScopedLogger bind(logger);
Review Comment:
Please catch exceptions inside each delete task and include them in the failure count. TaskGroup currently invokes the task without catching exceptions, and Arrow's worker loop does not catch them either, so a throwing delete can terminate the process instead of returning a Status. Please cover this with a throwing-task test and verify that the remaining files are still attempted.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4130050711
##########
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:
@zhjwpku Thanks! TaskGroup needs a caller-supplied Executor, which DeleteFiles can't receive, and ExpireSnapshots already retries it. So plain threads here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "zhjwpku (via GitHub)" <gi...@apache.org>.
zhjwpku commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4110740970
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -284,12 +284,18 @@ 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;
+ // Like Java's S3FileIO, keep going after a failure and report the count.
+ // Arrow's S3 DeleteFiles deletes one file at a time too.
+ size_t failed = 0;
for (const auto& file_location : file_locations) {
- locations_by_io[&FileIOForPath(file_location)].push_back(file_location);
+ if (auto status = FileIOForPath(file_location).DeleteFile(file_location);
Review Comment:
Should we use a thread pool to delete files concurrently? Sequential S3 deletes could be slow for large batches. Java's S3FileIO does that, so maybe we should pursue that too, but not a requirement for this PR.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(io): attempt every S3 delete and report how many failed [iceberg-cpp]
Posted by "plusplusjiajia (via GitHub)" <gi...@apache.org>.
plusplusjiajia commented on code in PR #966:
URL: https://github.com/apache/iceberg-cpp/pull/966#discussion_r4111392917
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -284,12 +284,18 @@ 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;
+ // Like Java's S3FileIO, keep going after a failure and report the count.
+ // Arrow's S3 DeleteFiles deletes one file at a time too.
+ size_t failed = 0;
for (const auto& file_location : file_locations) {
- locations_by_io[&FileIOForPath(file_location)].push_back(file_location);
+ if (auto status = FileIOForPath(file_location).DeleteFile(file_location);
Review Comment:
@zhjwpku Thanks! Done: deletes now run on up to s3.delete.num-threads threads (default: hardware threads), as in Java. Java also batches keys into DeleteObjects; Arrow has no public batch delete for S3, so each thread deletes one file at a time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org