diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index c981759ce..77f8f4934 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -17,11 +17,15 @@ * under the License. */ +#include +#include #include +#include #include #include #include #include +#include #include #include #include @@ -36,6 +40,7 @@ #include "iceberg/arrow/arrow_status_internal.h" #include "iceberg/arrow/s3/s3_properties.h" #include "iceberg/logging/log_macros.h" +#include "iceberg/logging/logger.h" #include "iceberg/util/macros.h" #include "iceberg/util/property_util.h" #include "iceberg/util/string_util.h" @@ -178,9 +183,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 default_properties) + std::unordered_map default_properties, + size_t delete_threads) : default_file_io_(std::move(arrow_fs)), - default_properties_(std::move(default_properties)) {} + default_properties_(std::move(default_properties)), + delete_threads_(delete_threads) {} Result> NewInputFile(std::string file_location) override; @@ -207,6 +214,7 @@ class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials { ArrowFileSystemFileIO default_file_io_; std::unordered_map default_properties_; + size_t delete_threads_; std::vector storage_credentials_; std::vector>> file_io_by_prefix_; @@ -284,12 +292,37 @@ Status ArrowS3FileIO::DeleteFile(const std::string& file_location) { } Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations) { - std::unordered_map> 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 next = 0; + std::atomic 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> helpers; + for (size_t i = 1; i < std::min(delete_threads_, file_locations.size()); ++i) { + helpers.push_back(std::async(std::launch::async, [&] { + ScopedLogger bind(logger); + delete_remaining(); + })); + } + delete_remaining(); + for (auto& helper : helpers) { + helper.get(); // Rethrows what a helper threw. } - for (auto& [file_io, locations] : locations_by_io) { - ICEBERG_RETURN_UNEXPECTED(file_io->DeleteFiles(locations)); + if (failed > 0) { + return IOError("Failed to delete {} of {} files", failed.load(), + file_locations.size()); } return {}; } @@ -300,7 +333,16 @@ Result> MakeS3FileIO( const std::unordered_map& properties) { // Uses default credentials if properties are empty. ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties)); - return std::make_unique(std::move(fs), properties); + // Java defaults to one delete thread per processor. + size_t delete_threads = std::max(1u, std::thread::hardware_concurrency()); + if (const auto* value = FindProperty(properties, S3Properties::kDeleteNumThreads); + value != nullptr) { + ICEBERG_ASSIGN_OR_RAISE(delete_threads, StringUtils::ParseNumber(*value)); + if (delete_threads == 0) { + return InvalidArgument(R"("{}" must be positive)", S3Properties::kDeleteNumThreads); + } + } + return std::make_unique(std::move(fs), properties, delete_threads); } Status FinalizeS3() { diff --git a/src/iceberg/arrow/s3/s3_properties.h b/src/iceberg/arrow/s3/s3_properties.h index 50dafa56f..bcc186583 100644 --- a/src/iceberg/arrow/s3/s3_properties.h +++ b/src/iceberg/arrow/s3/s3_properties.h @@ -54,6 +54,8 @@ struct S3Properties { static constexpr std::string_view kConnectTimeoutMs = "s3.connect-timeout-ms"; /// Socket timeout in milliseconds static constexpr std::string_view kSocketTimeoutMs = "s3.socket-timeout-ms"; + /// Threads DeleteFiles uses; defaults to the hardware thread count, as in Java + static constexpr std::string_view kDeleteNumThreads = "s3.delete.num-threads"; }; /// \brief URI schemes served by the Arrow S3 FileIO, lower-case. diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc index 7235ea367..6c9c289bd 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -245,6 +245,15 @@ TEST_F(ArrowS3FileIOTest, RejectsIncompleteStaticCredentials) { "S3 client access key ID and secret access key must be set")); } +TEST_F(ArrowS3FileIOTest, RejectsInvalidDeleteThreads) { + for (std::string_view threads : {"0", "-1", "many"}) { + SCOPED_TRACE(threads); + EXPECT_THAT(MakeS3FileIO({{std::string(S3Properties::kDeleteNumThreads), + std::string(threads)}}), + IsError(ErrorKind::kInvalidArgument)); + } +} + TEST_F(ArrowS3FileIOTest, ReadWrite) { if (!HasIntegrationEnv()) { GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable S3 IO test"; @@ -299,6 +308,56 @@ TEST_F(ArrowS3FileIOTest, LongestCredentialPrefix) { IsOk()); } +TEST_F(ArrowS3FileIOTest, DeleteFilesAttemptsEveryFile) { + if (!HasIntegrationEnv()) { + GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable S3 IO test"; + } + + auto properties = PropertiesFromEnv(); + if (properties.empty()) { + GTEST_SKIP() << "Set S3 properties to enable credential routing test"; + } + // A thread per file, so the deletes run concurrently. + properties[std::string(S3Properties::kDeleteNumThreads)] = "3"; + + auto io_res = MakeS3FileIO(properties); + ASSERT_THAT(io_res, IsOk()); + auto io = std::move(io_res).value(); + auto* credentialed = io->AsSupportsStorageCredentials(); + ASSERT_NE(credentialed, nullptr); + + const auto denied = ObjectUri("delete_denied/"); + const auto allowed = ObjectUri("delete_allowed/") + "only"; + const std::vector paths = {denied + "first", allowed, denied + "second"}; + for (const auto& path : paths) { + ASSERT_THAT(io->WriteFile(path, "payload"), IsOk()); + } + + // Deletes under `denied` fail; the one between them must still run. + auto bad_properties = properties; + for (const auto& [key, value] : BadS3Credentials()) { + bad_properties.insert_or_assign(key, value); + } + ASSERT_THAT(credentialed->SetStorageCredentials( + {{.prefix = denied, .config = std::move(bad_properties)}}), + IsOk()); + // Each failure reaches the caller's logger, whichever thread hit it. + auto logger = std::make_shared(); + { + ScopedLogger bind(logger); + EXPECT_THAT(io->DeleteFiles(paths), HasErrorMessage("Failed to delete 2 of 3 files")); + } + EXPECT_EQ(std::ranges::count_if( + logger->records(), + [](const LogMessage& record) { return record.level == LogLevel::kWarn; }), + 2); + EXPECT_FALSE(io->ReadFile(allowed, std::nullopt).has_value()); + + // The denied files remain: Arrow fails to delete a missing object. + ASSERT_THAT(credentialed->SetStorageCredentials({}), IsOk()); + EXPECT_THAT(io->DeleteFiles({paths[0], paths[2]}), IsOk()); +} + // The credential is vended under the oss spelling and the object addressed as // `s3://`, so they only meet through canonicalization — and every other path // to authentication is broken. (rest_arrow_file_io_test covers the mirrored