FileIO only deal with metadata files

Author: zhjwpku

cmake_modules/IcebergThirdpartyToolchain.cmake

@@ -70,7 +70,7 @@ function(resolve_arrow_dependency)
       ON
       CACHE BOOL "" FORCE)
   set(ARROW_FILESYSTEM
-      OFF
+      ON
       CACHE BOOL "" FORCE)
   set(ARROW_SIMD_LEVEL
       "NONE"

src/iceberg/CMakeLists.txt

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

src/iceberg/arrow/CMakeLists.txt

@@ -15,4 +15,6 @@
 # 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

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <arrow/result.h>
+#include <arrow/status.h>
+
+#include "iceberg/error.h"
+#include "iceberg/expected.h"
+
+namespace iceberg::arrow::internal {
+
+inline ErrorKind ToErrorKind(const ::arrow::Status& status) {
+  switch (status.code()) {
+    case ::arrow::StatusCode::IOError:
+      return ErrorKind::kIOError;
+    default:
+      return ErrorKind::kUnknownError;
+  }
+}
+
+#define ICEBERG_INTERNAL_ASSIGN_OR_RETURN_IMPL(result_name, lhs, rexpr, error_transform) \
+  auto&& result_name = (rexpr);                                                          \
+  if (!result_name.ok()) {                                                               \
+    return unexpected(                                                                   \
+        Error(error_transform(result_name.status()), result_name.status().ToString()));  \
+  }                                                                                      \
+  lhs = std::move(result_name).ValueOrDie();
+
+#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, \
+      internal::ToErrorKind)
+
+#define ICEBERG_INTERNAL_RETURN_NOT_OK(expr)                                        \
+  do {                                                                              \
+    auto&& _status = (expr);                                                        \
+    if (!_status.ok()) {                                                            \
+      return unexpected(Error(internal::ToErrorKind(_status), _status.ToString())); \
+    }                                                                               \
+  } while (0)
+
+}  // namespace iceberg::arrow::internal

test/io/CMakeLists.txt renamed to src/iceberg/arrow/io/CMakeLists.txt

@@ -15,10 +15,4 @@
 # specific language governing permissions and limitations
 # under the License.
 
-if(ICEBERG_IO)
-  add_executable(fs_file_io_test)
-  target_sources(fs_file_io_test PRIVATE fs_file_io_test.cc)
-  target_link_libraries(fs_file_io_test PRIVATE iceberg_io_static GTest::gtest_main)
-  target_include_directories(fs_file_io_test PRIVATE "${ICEBERG_INCLUDES}")
-  add_test(NAME fs_file_io_test COMMAND fs_file_io_test)
-endif()
+iceberg_install_all_headers(iceberg/arrow/io)

src/iceberg/arrow/io/local_file_io.cc

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/arrow/io/local_file_io.h"
+
+#include <filesystem>
+
+#include <arrow/filesystem/localfs.h>
+
+#include "iceberg/arrow/arrow_error_transform_internal.h"
+
+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) {
+  // 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_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])));
+
+  return content;
+}
+
+/// \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, ""));
+  }
+  ICEBERG_INTERNAL_ASSIGN_OR_RETURN(auto file,
+                                    local_fs_->OpenOutputStream(file_location));
+  ICEBERG_INTERNAL_RETURN_NOT_OK(file->Write(content.data(), content.size()));
+  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));
+  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

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include <arrow/filesystem/localfs.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 {
+ public:
+  explicit LocalFileIO(std::shared_ptr<::arrow::fs::LocalFileSystem>& local_fs)
+      : local_fs_(local_fs) {}
+
+  ~LocalFileIO() override = default;
+
+  /// \brief Read the content of the file at the given location.
+  expected<std::string, Error> ReadFile(const std::string& file_location,
+                                        std::optional<size_t> length) override;
+
+  /// \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;
+
+  /// \brief Delete a file at the given location.
+  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_;
+};
+
+}  // namespace iceberg::arrow::io

src/iceberg/error.h

@@ -30,16 +30,22 @@ namespace iceberg {
 enum class ErrorKind {
   kNoSuchNamespace,
   kAlreadyExists,
+  kNoSuchFile,
   kNoSuchTable,
   kCommitStateUnknown,
   kInvalidSchema,
   kInvalidArgument,
+  kIOError,
+  kUnknownError,
 };
 
 /// \brief Error with a kind and a message.
 struct ICEBERG_EXPORT [[nodiscard]] Error {
   ErrorKind kind;
   std::string message;
+
+  explicit Error(ErrorKind kind, std::string message)
+      : kind(kind), message(std::move(message)) {};
 };
 
 }  // namespace iceberg

src/iceberg/file_io.h

@@ -19,92 +19,51 @@
 
 #pragma once
 
-#include <memory>
+#include <optional>
 #include <string>
+#include <string_view>
 
+#include "iceberg/error.h"
+#include "iceberg/expected.h"
 #include "iceberg/iceberg_export.h"
-#include "iceberg/reader.h"
-#include "iceberg/writer.h"
 
 namespace iceberg {
 
-/// \brief An interface used to read input files using Reader and AsyncReader
-class ICEBERG_EXPORT InputFile {
- public:
-  explicit InputFile(std::string location) : location_(std::move(location)) {}
-
-  virtual ~InputFile() = default;
-
-  /// \brief Checks whether the file exists.
-  virtual bool exists() const = 0;
-  /// \brief Returns the total length of the file, in bytes.
-  virtual int64_t getLength() const = 0;
-
-  /// \brief Get a Reader instance to read bytes from the file.
-  virtual std::unique_ptr<Reader> newReader() = 0;
-
-  /// \brief Get the file location
-  const std::string& location() const { return location_; }
-
- protected:
-  std::string location_;
-};
-
-/// \brief An interface used to write output files using Writer and AsyncWriter
-class ICEBERG_EXPORT OutputFile {
- public:
-  explicit OutputFile(std::string location) : location_(std::move(location)) {}
-
-  virtual ~OutputFile() = default;
-
-  /// \brief Create the file.
-  ///
-  /// If the file exists, or an error will be thrown.
-  virtual void create() = 0;
-
-  /// \brief Get a Writer instance to write bytes to the file.
-  virtual std::unique_ptr<Writer> newWriter() = 0;
-
-  /// \brief Get the file location
-  const std::string& location() const { return location_; }
-
-  /// \brief Return an InputFile for the location of this OutputFile.
-  virtual std::shared_ptr<InputFile> toInputFile() const = 0;
-
- protected:
-  std::string location_;
-};
-
-/// \brief Pluggable module for reading, writing, and deleting files.
+/// \brief Pluggable module for reading, writing, and deleting metadata files.
 ///
-/// Both table metadata files and data files can be written and read by this module.
+/// 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.
 class ICEBERG_EXPORT FileIO {
  public:
-  explicit FileIO(std::string name) : name_(std::move(name)) {}
-
+  FileIO() = default;
   virtual ~FileIO() = default;
 
-  /// \brief Get an InputFile
+  /// \brief Read the content of the file at the given location.
   ///
-  /// Get a InputFile instance to read bytes from the file at the given location.
-  virtual std::shared_ptr<InputFile> newInputFile(const std::string& location) = 0;
-
-  /// \brief Get an OutputFile
+  /// \param file_location The location of the file to read.
+  /// \param length The number of bytes to read. Some object storage need to specify
+  /// the length to read, e.g. S3 `GetObject` has a Range parameter.
+  /// \return The content of the file if the read succeeded, an error code if the read
+  /// failed.
+  virtual expected<std::string, Error> ReadFile(const std::string& file_location,
+                                                std::optional<size_t> length) = 0;
+
+  /// \brief Write the given content to the file at the given location.
   ///
-  /// Get a OutputFile instance to write bytes to the file at the given location.
-  virtual std::shared_ptr<OutputFile> newOutputFile(const std::string& location) = 0;
-
-  /// \brief Delete file
+  /// \param file_location The location of the file to write.
+  /// \param content The content to write to the file.
+  /// \param overwrite If true, overwrite the file if it exists. If false, fail if the
+  /// 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) = 0;
+
+  /// \brief Delete a file at the given location.
   ///
-  /// Delete the file at the given location.
-  virtual void DeleteFile(const std::string& location) = 0;
-  void DeleteFile(const InputFile& ifile) { return DeleteFile(ifile.location()); }
-  void DeleteFile(const OutputFile& ofile) { return DeleteFile(ofile.location()); }
-
-  const std::string& name() const { return name_; }
-
- protected:
-  std::string name_;
+  /// \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;
 };
 
 }  // namespace iceberg