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

Author: slfan1989

src/iceberg/file_io.h

@@ -151,7 +151,7 @@ class ICEBERG_EXPORT FileIO {
   ///
   /// \param file_location The location of the file to delete.
   /// \return void if the delete succeeded, an error code if the delete failed.
-  virtual Status DeleteFile(const std::string&) {
+  virtual Status DeleteFile(const std::string& file_location) {
     return NotImplemented("DeleteFile not implemented");
   }
 

src/iceberg/test/file_io_test.cc

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

src/iceberg/update/expire_snapshots.cc

@@ -75,26 +75,22 @@ class FileCleanupStrategy {
                             CleanupLevel level) = 0;
 
  protected:
-  /// \brief Delete a single file
-  void DeleteFile(const std::string& path) {
+  /// \brief Delete files at the given locations.
+  void DeleteFiles(const std::unordered_set<std::string>& paths) {
     try {
       if (delete_func_) {
-        delete_func_(path);
+        for (const auto& path : paths) {
+          delete_func_(path);
+        }
       } else {
-        std::ignore = file_io_->DeleteFile(path);
+        std::vector<std::string> path_list(paths.begin(), paths.end());
+        std::ignore = file_io_->DeleteFiles(path_list);
       }
     } catch (...) {
       /// TODO(shangxinli): add retry
     }
   }
 
-  /// TODO(shangxinli): Add bulk deletion
-  void DeleteFiles(const std::unordered_set<std::string>& paths) {
-    for (const auto& path : paths) {
-      DeleteFile(path);
-    }
-  }
-
   bool HasAnyStatisticsFiles(const TableMetadata& metadata) const {
     return !metadata.statistics.empty() || !metadata.partition_statistics.empty();
   }