fix review comments

Signed-off-by: Junwang Zhao <[email protected]>

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -56,7 +56,7 @@ install(FILES ${CMAKE_CURRENT_BINARY_DIR}/iceberg_export.h
         DESTINATION ${ICEBERG_INSTALL_INCLUDEDIR}/iceberg)
 
 if(ICEBERG_BUILD_BUNDLE)
-  set(ICEBERG_BUNDLE_SOURCES arrow/demo_arrow.cc arrow/io/local_file_io.cc
+  set(ICEBERG_BUNDLE_SOURCES arrow/demo_arrow.cc arrow/io/arrow_fs_file_io.cc
                              avro/demo_avro.cc)
 
   # Libraries to link with exported libiceberg_bundle.{so,a}.

src/iceberg/arrow/arrow_error_transform_internal.h

@@ -27,6 +27,8 @@
 
 namespace iceberg::arrow::internal {
 
+namespace {
+
 inline ErrorKind ToErrorKind(const ::arrow::Status& status) {
   switch (status.code()) {
     case ::arrow::StatusCode::IOError:
@@ -44,6 +46,8 @@ inline ErrorKind ToErrorKind(const ::arrow::Status& status) {
   }                                                                                      \
   lhs = std::move(result_name).ValueOrDie();
 
+}  // namespace
+
 #define ICEBERG_INTERNAL_ASSIGN_OR_RETURN(lhs, rexpr)                       \
   ICEBERG_INTERNAL_ASSIGN_OR_RETURN_IMPL(                                   \
       ARROW_ASSIGN_OR_RAISE_NAME(_error_or_value, __COUNTER__), lhs, rexpr, \

src/iceberg/arrow/io/local_file_io.cc renamed to src/iceberg/arrow/io/arrow_fs_file_io.cc

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-#include "iceberg/arrow/io/local_file_io.h"
+#include "iceberg/arrow/io/arrow_fs_file_io.h"
 
 #include <filesystem>
 
@@ -28,14 +28,14 @@
 namespace iceberg::arrow::io {
 
 /// \brief Read the content of the file at the given location.
-expected<std::string, Error> LocalFileIO::ReadFile(const std::string& file_location,
-                                                   std::optional<size_t> length) {
+expected<std::string, Error> ArrowFileSystemFileIO::ReadFile(
+    const std::string& file_location, std::optional<size_t> length) {
   // We don't support reading a file with a specific length.
   if (length.has_value()) {
     return unexpected(Error(ErrorKind::kInvalidArgument, "Length is not supported"));
   }
   std::string content;
-  ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file, local_fs_->OpenInputFile(file_location));
+  ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file, arrow_fs_->OpenInputFile(file_location));
   ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file_size, file->GetSize());
 
   content.resize(file_size);
@@ -47,31 +47,28 @@ expected<std::string, Error> LocalFileIO::ReadFile(const std::string& file_locat
 }
 
 /// \brief Write the given content to the file at the given location.
-expected<void, Error> LocalFileIO::WriteFile(const std::string& file_location,
-                                             std::string_view content, bool overwrite) {
-  if (!overwrite && FileExists(file_location)) {
-    return unexpected(
-        Error(ErrorKind::kAlreadyExists, std::format("File {} exists", file_location)));
-  }
+expected<void, Error> ArrowFileSystemFileIO::WriteFile(const std::string& file_location,
+                                                       std::string_view content,
+                                                       bool overwrite) {
+  // auto file_info = arrow_fs_->GetFileInfo(file_location);
+  // if (file_info.status().ok() && !overwrite) {
+  //   return unexpected(
+  //       Error(ErrorKind::kAlreadyExists, std::format("File {} exists",
+  //       file_location)));
+  // }
   ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file,
-                                    local_fs_->OpenOutputStream(file_location));
+                                    arrow_fs_->OpenOutputStream(file_location));
   ICEBERG_INTERNAL_RETURN_NOT_OK(file->Write(content.data(), content.size()));
+  ICEBERG_INTERNAL_RETURN_NOT_OK(file->Flush());
+  ICEBERG_INTERNAL_RETURN_NOT_OK(file->Close());
   return {};
 }
 
 /// \brief Delete a file at the given location.
-expected<void, Error> LocalFileIO::DeleteFile(const std::string& file_location) {
-  if (!FileExists(file_location)) {
-    return unexpected(Error(ErrorKind::kNoSuchFile,
-                            std::format("File {} does not exist", file_location)));
-  }
-  ICEBERG_INTERNAL_RETURN_NOT_OK(local_fs_->DeleteFile(file_location));
+expected<void, Error> ArrowFileSystemFileIO::DeleteFile(
+    const std::string& file_location) {
+  ICEBERG_INTERNAL_RETURN_NOT_OK(arrow_fs_->DeleteFile(file_location));
   return {};
 }
 
-bool LocalFileIO::FileExists(const std::string& location) {
-  // ::arrow::fs::LocalFileSystem does not have a exists method.
-  return std::filesystem::exists(location);
-}
-
 }  // namespace iceberg::arrow::io

src/iceberg/arrow/io/local_file_io.h renamed to src/iceberg/arrow/io/arrow_fs_file_io.h

@@ -21,20 +21,20 @@
 
 #include <memory>
 
-#include <arrow/filesystem/localfs.h>
+#include <arrow/filesystem/filesystem.h>
 
 #include "iceberg/file_io.h"
 #include "iceberg/iceberg_bundle_export.h"
 
 namespace iceberg::arrow::io {
 
-/// \brief A concrete implementation of FileIO for file system.
-class ICEBERG_BUNDLE_EXPORT LocalFileIO : public FileIO {
+/// \brief A concrete implementation of FileIO for Arrow file system.
+class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
  public:
-  explicit LocalFileIO(std::shared_ptr<::arrow::fs::LocalFileSystem>& local_fs)
-      : local_fs_(local_fs) {}
+  explicit ArrowFileSystemFileIO(std::shared_ptr<::arrow::fs::FileSystem> arrow_fs)
+      : arrow_fs_(std::move(arrow_fs)) {}
 
-  ~LocalFileIO() override = default;
+  ~ArrowFileSystemFileIO() override = default;
 
   /// \brief Read the content of the file at the given location.
   expected<std::string, Error> ReadFile(const std::string& file_location,
@@ -48,10 +48,7 @@ class ICEBERG_BUNDLE_EXPORT LocalFileIO : public FileIO {
   expected<void, Error> DeleteFile(const std::string& file_location) override;
 
  private:
-  /// \brief Check if a file exists
-  bool FileExists(const std::string& location);
-
-  std::shared_ptr<::arrow::fs::LocalFileSystem>& local_fs_;
+  std::shared_ptr<::arrow::fs::FileSystem> arrow_fs_;
 };
 
 }  // namespace iceberg::arrow::io

src/iceberg/error.h

@@ -30,7 +30,6 @@ namespace iceberg {
 enum class ErrorKind {
   kNoSuchNamespace,
   kAlreadyExists,
-  kNoSuchFile,
   kNoSuchTable,
   kCommitStateUnknown,
   kInvalidArgument,

src/iceberg/file_io.h

@@ -29,7 +29,7 @@
 
 namespace iceberg {
 
-/// \brief Pluggable module for reading, writing, and deleting metadata files.
+/// \brief Pluggable module for reading, writing, and deleting files.
 ///
 /// This module only handle metadata files, not data files. The metadata files
 /// are typically small and are used to store schema, partition information,

test/CMakeLists.txt

@@ -44,9 +44,9 @@ if(ICEBERG_BUILD_BUNDLE)
   target_link_libraries(arrow_test PRIVATE iceberg_bundle_static GTest::gtest_main)
   add_test(NAME arrow_test COMMAND arrow_test)
 
-  add_executable(local_file_io_test)
-  target_sources(local_file_io_test PRIVATE local_file_io_test.cc)
-  target_link_libraries(local_file_io_test PRIVATE iceberg_bundle_static
-                                                   GTest::gtest_main)
-  add_test(NAME local_file_io_test COMMAND local_file_io_test)
+  add_executable(arrow_fs_file_io_test)
+  target_sources(arrow_fs_file_io_test PRIVATE arrow_fs_file_io_test.cc)
+  target_link_libraries(arrow_fs_file_io_test PRIVATE iceberg_bundle_static
+                                                      GTest::gtest_main)
+  add_test(NAME arrow_fs_file_io_test COMMAND arrow_fs_file_io_test)
 endif()

test/local_file_io_test.cc renamed to test/arrow_fs_file_io_test.cc

@@ -17,17 +17,18 @@
  * under the License.
  */
 
-#include "iceberg/arrow/io/local_file_io.h"
+#include "iceberg/arrow/io/arrow_fs_file_io.h"
 
 #include <filesystem>
 
+#include <arrow/filesystem/localfs.h>
 #include <gtest/gtest.h>
 
 class LocalFileIOTest : public testing::Test {
  protected:
   void SetUp() override {
     local_fs_ = std::make_shared<arrow::fs::LocalFileSystem>();
-    file_io_ = std::make_shared<iceberg::arrow::io::LocalFileIO>(local_fs_);
+    file_io_ = std::make_shared<iceberg::arrow::io::ArrowFileSystemFileIO>(local_fs_);
   }
 
   std::shared_ptr<arrow::fs::LocalFileSystem> local_fs_;
@@ -54,5 +55,5 @@ TEST_F(LocalFileIOTest, DeleteFile) {
   EXPECT_TRUE(del_res.has_value());
 
   del_res = file_io_->DeleteFile(tmpfile.string());
-  EXPECT_EQ(del_res.error().kind, iceberg::ErrorKind::kNoSuchFile);
+  EXPECT_EQ(del_res.error().kind, iceberg::ErrorKind::kIOError);
 }