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