feat(io): add bulk delete API to FileIO.

Author: slfan1989

src/iceberg/arrow/arrow_io.cc

@@ -22,6 +22,7 @@
 #include <limits>
 #include <mutex>
 #include <optional>
+#include <vector>
 
 #include <arrow/buffer.h>
 #include <arrow/filesystem/localfs.h>
@@ -568,6 +569,18 @@ Status ArrowFileSystemFileIO::DeleteFile(const std::string& file_location) {
   return {};
 }
 
+Status ArrowFileSystemFileIO::DeleteFiles(
+    const std::vector<std::string>& file_locations) {
+  std::vector<std::string> paths;
+  paths.reserve(file_locations.size());
+  for (const auto& file_location : file_locations) {
+    ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
+    paths.push_back(std::move(path));
+  }
+  ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFiles(paths));
+  return {};
+}
+
 std::unique_ptr<FileIO> ArrowFileSystemFileIO::MakeMockFileIO() {
   return std::make_unique<ArrowFileSystemFileIO>(
       std::make_shared<::arrow::fs::internal::MockFileSystem>(

src/iceberg/arrow/arrow_io_internal.h

@@ -23,6 +23,7 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <vector>
 
 #include <arrow/filesystem/type_fwd.h>
 #include <arrow/io/type_fwd.h>
@@ -77,6 +78,9 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
   /// \brief Delete a file at the given location.
   Status DeleteFile(const std::string& file_location) override;
 
+  /// \brief Delete files at the given locations.
+  Status DeleteFiles(const std::vector<std::string>& file_locations) override;
+
   /// \brief Get the Arrow file system.
   const std::shared_ptr<::arrow::fs::FileSystem>& fs() const { return arrow_fs_; }
 

src/iceberg/file_io.cc

@@ -100,4 +100,11 @@ Status FileIO::WriteFile(const std::string& file_location, std::string_view cont
   return FinishWithCloseStatus(std::move(status), stream->Close());
 }
 
+Status FileIO::DeleteFiles(const std::vector<std::string>& file_locations) {
+  for (const auto& file_location : file_locations) {
+    ICEBERG_RETURN_UNEXPECTED(DeleteFile(file_location));
+  }
+  return {};
+}
+
 }  // namespace iceberg

src/iceberg/file_io.h

@@ -26,6 +26,7 @@
 #include <span>
 #include <string>
 #include <string_view>
+#include <vector>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
@@ -163,15 +164,7 @@ class ICEBERG_EXPORT FileIO {
   ///
   /// \param file_locations The locations of the files to delete.
   /// \return void if all deletes succeeded, an error code if any delete failed.
-  virtual Status DeleteFiles(std::span<const std::string> file_locations) {
-    for (const auto& file_location : file_locations) {
-      auto status = DeleteFile(file_location);
-      if (!status.has_value()) {
-        return status;
-      }
-    }
-    return {};
-  }
+  virtual Status DeleteFiles(const std::vector<std::string>& file_locations);
 };
 
 }  // namespace iceberg

src/iceberg/test/arrow_io_test.cc

@@ -21,6 +21,7 @@
 #include <array>
 #include <memory>
 #include <string>
+#include <vector>
 
 #include <arrow/filesystem/localfs.h>
 #include <arrow/result.h>
@@ -341,6 +342,20 @@ TEST_F(LocalFileIOTest, DeleteFile) {
   EXPECT_THAT(del_res, HasErrorMessage("Cannot delete file"));
 }
 
+TEST_F(LocalFileIOTest, DeleteFiles) {
+  auto first_path = CreateNewTempFilePath();
+  auto second_path = CreateNewTempFilePath();
+  ASSERT_THAT(file_io_->WriteFile(first_path, "hello"), IsOk());
+  ASSERT_THAT(file_io_->WriteFile(second_path, "world"), IsOk());
+
+  std::vector<std::string> paths = {first_path, second_path};
+  EXPECT_THAT(file_io_->DeleteFiles(paths), IsOk());
+
+  EXPECT_THAT(file_io_->ReadFile(first_path, std::nullopt), IsError(ErrorKind::kIOError));
+  EXPECT_THAT(file_io_->ReadFile(second_path, std::nullopt),
+              IsError(ErrorKind::kIOError));
+}
+
 void VerifyReadFullyReadsFromAbsolutePosition(const std::shared_ptr<FileIO>& file_io,
                                               const std::string& path) {
   ASSERT_THAT(file_io->WriteFile(path, "abcdef"), IsOk());

src/iceberg/test/file_io_test.cc

@@ -27,17 +27,18 @@
 
 #include "iceberg/test/matchers.h"
 
+namespace iceberg {
 namespace {
 
-class RecordingFileIO : public iceberg::FileIO {
+class RecordingFileIO : public FileIO {
  public:
   explicit RecordingFileIO(std::string failure_path = "")
       : failure_path_(std::move(failure_path)) {}
 
-  iceberg::Status DeleteFile(const std::string& file_location) override {
+  Status DeleteFile(const std::string& file_location) override {
     deleted_paths.push_back(file_location);
     if (file_location == failure_path_) {
-      return iceberg::IOError("failed to delete {}", file_location);
+      return IOError("failed to delete {}", file_location);
     }
     return {};
   }
@@ -54,7 +55,7 @@ TEST(FileIOTest, DeleteFilesFallsBackToDeleteFileForEachPath) {
   RecordingFileIO file_io;
   std::vector<std::string> paths = {"file-a.avro", "file-b.avro"};
 
-  EXPECT_THAT(file_io.DeleteFiles(paths), iceberg::IsOk());
+  EXPECT_THAT(file_io.DeleteFiles(paths), IsOk());
   EXPECT_THAT(file_io.deleted_paths,
               ::testing::ElementsAre("file-a.avro", "file-b.avro"));
 }
@@ -65,8 +66,10 @@ TEST(FileIOTest, DeleteFilesReturnsFirstDeleteFileError) {
 
   auto status = file_io.DeleteFiles(paths);
 
-  EXPECT_THAT(status, iceberg::IsError(iceberg::ErrorKind::kIOError));
-  EXPECT_THAT(status, iceberg::HasErrorMessage("failed to delete file-b.avro"));
+  EXPECT_THAT(status, IsError(ErrorKind::kIOError));
+  EXPECT_THAT(status, HasErrorMessage("failed to delete file-b.avro"));
   EXPECT_THAT(file_io.deleted_paths,
               ::testing::ElementsAre("file-a.avro", "file-b.avro"));
 }
+
+}  // namespace iceberg