fix more review comments

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

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -57,7 +57,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/arrow_fs_file_io.cc
+  set(ICEBERG_BUNDLE_SOURCES arrow/demo_arrow.cc arrow/arrow_fs_file_io.cc
                              avro/demo_avro.cc)
 
   # Libraries to link with exported libiceberg_bundle.{so,a}.

src/iceberg/arrow/CMakeLists.txt

@@ -15,6 +15,4 @@
 # specific language governing permissions and limitations
 # under the License.
 
-add_subdirectory(io)
-
 iceberg_install_all_headers(iceberg/arrow)

src/iceberg/arrow/arrow_error_transform_internal.h

@@ -36,20 +36,20 @@ inline ErrorKind ToErrorKind(const ::arrow::Status& status) {
   }
 }
 
-#define ICEBERG_INTERNAL_ASSIGN_OR_RETURN_IMPL(result_name, lhs, rexpr, error_transform) \
-  auto&& result_name = (rexpr);                                                          \
-  if (!result_name.ok()) {                                                               \
-    return unexpected<Error>{{.kind = error_transform(result_name.status()),             \
-                              .message = result_name.status().ToString()}};              \
-  }                                                                                      \
+#define ICEBERG_ARROW_ASSIGN_OR_RETURN_IMPL(result_name, lhs, rexpr, error_transform) \
+  auto&& result_name = (rexpr);                                                       \
+  if (!result_name.ok()) {                                                            \
+    return unexpected<Error>{{.kind = error_transform(result_name.status()),          \
+                              .message = result_name.status().ToString()}};           \
+  }                                                                                   \
   lhs = std::move(result_name).ValueOrDie();
 
-#define ICEBERG_INTERNAL_ASSIGN_OR_RETURN(lhs, rexpr)                       \
-  ICEBERG_INTERNAL_ASSIGN_OR_RETURN_IMPL(                                   \
+#define ICEBERG_ARROW_ASSIGN_OR_RETURN(lhs, rexpr)                          \
+  ICEBERG_ARROW_ASSIGN_OR_RETURN_IMPL(                                      \
       ARROW_ASSIGN_OR_RAISE_NAME(_error_or_value, __COUNTER__), lhs, rexpr, \
       internal::ToErrorKind)
 
-#define ICEBERG_INTERNAL_RETURN_NOT_OK(expr)                                        \
+#define ICEBERG_ARROW_RETURN_NOT_OK(expr)                                           \
   do {                                                                              \
     auto&& _status = (expr);                                                        \
     if (!_status.ok()) {                                                            \

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

@@ -17,56 +17,55 @@
  * under the License.
  */
 
-#include "iceberg/arrow/io/arrow_fs_file_io.h"
-
-#include <filesystem>
+#include "iceberg/arrow/arrow_fs_file_io.h"
 
 #include <arrow/filesystem/localfs.h>
 
 #include "iceberg/arrow/arrow_error_transform_internal.h"
 
-namespace iceberg::arrow::io {
+namespace iceberg::arrow {
 
 /// \brief Read the content of the file at the given location.
 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.
+  ::arrow::fs::FileInfo file_info(file_location);
   if (length.has_value()) {
-    return unexpected(Error(ErrorKind::kInvalidArgument, "Length is not supported"));
+    file_info.set_size(length.value());
   }
   std::string content;
-  ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file, arrow_fs_->OpenInputFile(file_location));
-  ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file_size, file->GetSize());
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, arrow_fs_->OpenInputFile(file_info));
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file_size, file->GetSize());
 
   content.resize(file_size);
-  ICEBERG_INTERNAL_ASSIGN_OR_RETURN(
-      auto read_length,
-      file->ReadAt(0, file_size, reinterpret_cast<uint8_t*>(&content[0])));
+  size_t remain = file_size;
+  size_t offset = 0;
+  while (remain > 0) {
+    size_t read_length = std::min(remain, static_cast<size_t>(1024 * 1024));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(
+        auto read_bytes,
+        file->Read(read_length, reinterpret_cast<uint8_t*>(&content[offset])));
+    remain -= read_bytes;
+    offset += read_bytes;
+  }
 
   return content;
 }
 
 /// \brief Write the given content to the file at the given location.
 expected<void, Error> ArrowFileSystemFileIO::WriteFile(const std::string& file_location,
-                                                       std::string_view content,
-                                                       bool overwrite) {
-  auto exists = arrow_fs_->OpenInputFile(file_location).ok();
-  if (!overwrite && exists) {
-    return unexpected(Error(ErrorKind::kAlreadyExists, "File already exists"));
-  }
-  ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file,
-                                    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());
+                                                       std::string_view content) {
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, arrow_fs_->OpenOutputStream(file_location));
+  ICEBERG_ARROW_RETURN_NOT_OK(file->Write(content.data(), content.size()));
+  ICEBERG_ARROW_RETURN_NOT_OK(file->Flush());
+  ICEBERG_ARROW_RETURN_NOT_OK(file->Close());
   return {};
 }
 
 /// \brief Delete a file at the given location.
 expected<void, Error> ArrowFileSystemFileIO::DeleteFile(
     const std::string& file_location) {
-  ICEBERG_INTERNAL_RETURN_NOT_OK(arrow_fs_->DeleteFile(file_location));
+  ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFile(file_location));
   return {};
 }
 
-}  // namespace iceberg::arrow::io
+}  // namespace iceberg::arrow

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

@@ -26,7 +26,7 @@
 #include "iceberg/file_io.h"
 #include "iceberg/iceberg_bundle_export.h"
 
-namespace iceberg::arrow::io {
+namespace iceberg::arrow {
 
 /// \brief A concrete implementation of FileIO for Arrow file system.
 class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
@@ -42,7 +42,7 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
 
   /// \brief Write the given content to the file at the given location.
   expected<void, Error> WriteFile(const std::string& file_location,
-                                  std::string_view content, bool overwrite) override;
+                                  std::string_view content) override;
 
   /// \brief Delete a file at the given location.
   expected<void, Error> DeleteFile(const std::string& file_location) override;
@@ -51,4 +51,4 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
   std::shared_ptr<::arrow::fs::FileSystem> arrow_fs_;
 };
 
-}  // namespace iceberg::arrow::io
+}  // namespace iceberg::arrow

src/iceberg/arrow/io/CMakeLists.txt

@@ -34,6 +34,10 @@ namespace iceberg {
 /// This module only handle metadata files, not data files. The metadata files
 /// are typically small and are used to store schema, partition information,
 /// and other metadata about the table.
+///
+/// Note that these functions are not atomic. For example, if a write fails,
+/// the file may be partially written. Implementations should be careful to
+/// avoid corrupting metadata files.
 class ICEBERG_EXPORT FileIO {
  public:
   FileIO() = default;
@@ -48,10 +52,7 @@ class ICEBERG_EXPORT FileIO {
   /// failed.
   virtual expected<std::string, Error> ReadFile(const std::string& file_location,
                                                 std::optional<size_t> length) {
-    // The following line is to avoid Windows linker error LNK2019.
-    // If this function is defined as pure virtual function, the `expected<std::string,
-    // Error>` and `unexpected<Error>` will not be instantiated and exported in
-    // libiceberg.
+    // We provide a default implementation to avoid Windows linker error LNK2019.
     return unexpected<Error>{
         {.kind = ErrorKind::kNotImplemented, .message = "ReadFile not implemented"}};
   }
@@ -64,10 +65,7 @@ class ICEBERG_EXPORT FileIO {
   /// file exists.
   /// \return void if the write succeeded, an error code if the write failed.
   virtual expected<void, Error> WriteFile(const std::string& file_location,
-                                          std::string_view content, bool overwrite) {
-    // The following line is to avoid Windows linker error LNK2019.
-    // If this function is defined as pure virtual function, the `expected<void, Error>`
-    // will not be instantiated and exported in libiceberg.
+                                          std::string_view content) {
     return unexpected<Error>{
         {.kind = ErrorKind::kNotImplemented, .message = "WriteFile not implemented"}};
   }
@@ -76,7 +74,10 @@ 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 expected<void, Error> DeleteFile(const std::string& file_location) = 0;
+  virtual expected<void, Error> DeleteFile(const std::string& file_location) {
+    return unexpected<Error>{
+        {.kind = ErrorKind::kNotImplemented, .message = "DeleteFile not implemented"}};
+  }
 };
 
 }  // namespace iceberg

src/iceberg/file_io.h

@@ -41,14 +41,8 @@ if(ICEBERG_BUILD_BUNDLE)
   add_test(NAME avro_test COMMAND avro_test)
 
   add_executable(arrow_test)
-  target_sources(arrow_test PRIVATE arrow_test.cc)
+  target_sources(arrow_test PRIVATE arrow_test.cc arrow_fs_file_io_test.cc)
   target_link_libraries(arrow_test PRIVATE iceberg_bundle_static GTest::gtest_main
                                            GTest::gmock)
   add_test(NAME arrow_test COMMAND arrow_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 GTest::gmock)
-  add_test(NAME arrow_fs_file_io_test COMMAND arrow_fs_file_io_test)
 endif()

test/CMakeLists.txt

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-#include "iceberg/arrow/io/arrow_fs_file_io.h"
+#include "iceberg/arrow/arrow_fs_file_io.h"
 
 #include <filesystem>
 
@@ -32,7 +32,7 @@ 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::ArrowFileSystemFileIO>(local_fs_);
+    file_io_ = std::make_shared<iceberg::arrow::ArrowFileSystemFileIO>(local_fs_);
   }
 
   std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs_;
@@ -41,20 +41,16 @@ class LocalFileIOTest : public testing::Test {
 };
 
 TEST_F(LocalFileIOTest, ReadWriteFile) {
-  auto read_res = file_io_->ReadFile(tmpfile.string(), 1024);
-  EXPECT_THAT(read_res, IsError(ErrorKind::kInvalidArgument));
-  EXPECT_THAT(read_res, HasErrorMessage("Length is not supported"));
-
-  read_res = file_io_->ReadFile(tmpfile.string(), std::nullopt);
+  auto read_res = file_io_->ReadFile(tmpfile.string(), std::nullopt);
   EXPECT_THAT(read_res, IsError(ErrorKind::kIOError));
   EXPECT_THAT(read_res, HasErrorMessage("No such file or directory"));
 
-  auto write_res = file_io_->WriteFile(tmpfile.string(), "hello world", false);
+  auto write_res = file_io_->WriteFile(tmpfile.string(), "hello world");
   EXPECT_THAT(write_res, IsOk());
 
   read_res = file_io_->ReadFile(tmpfile.string(), std::nullopt);
   EXPECT_THAT(read_res, IsOk());
-  EXPECT_EQ(read_res.value(), "hello world");
+  EXPECT_THAT(read_res, HasValue(::testing::Eq("hello world")));
 }
 
 TEST_F(LocalFileIOTest, DeleteFile) {