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: [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]
