feat: add file_io and local fs impl

add reader and writer interfaces

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

Author: zhjwpku

CMakeLists.txt

@@ -38,6 +38,7 @@ option(ICEBERG_BUILD_SHARED "Build shared library" OFF)
 option(ICEBERG_BUILD_TESTS "Build tests" ON)
 option(ICEBERG_ARROW "Build Arrow" ON)
 option(ICEBERG_AVRO "Build Avro" ON)
+option(ICEBERG_IO "Build IO" ON)
 
 include(GNUInstallDirs)
 include(FetchContent)

src/iceberg/CMakeLists.txt

@@ -31,5 +31,6 @@ install(FILES ${CMAKE_CURRENT_BINARY_DIR}/iceberg_export.h
 add_subdirectory(arrow)
 add_subdirectory(avro)
 add_subdirectory(puffin)
+add_subdirectory(io)
 
 iceberg_install_cmake_package(Iceberg iceberg_targets)

src/iceberg/file_io.h

@@ -0,0 +1,113 @@
+/*
+ * 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 <string>
+#include <unordered_map>
+
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+// Forward declarations
+class Reader;
+class Writer;
+
+/// \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.
+///
+/// Both table metadata files and data files can be written and read by this module.
+class ICEBERG_EXPORT FileIO {
+ public:
+  explicit FileIO(std::string name) : name_(std::move(name)) {}
+
+  virtual ~FileIO() = default;
+
+  /// \brief Get an InputFile
+  ///
+  /// 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
+  ///
+  /// 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
+  ///
+  /// 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_;
+};
+
+}  // namespace iceberg

src/iceberg/io/CMakeLists.txt

@@ -0,0 +1,31 @@
+# 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.
+
+if(NOT ICEBERG_IO)
+  return()
+endif()
+
+set(ICEBERG_IO_SOURCES fs_file_io.cc)
+set(ICEBERG_IO_INCLUDES "${ICEBERG_INCLUDES}")
+
+add_iceberg_lib(iceberg_io
+                SOURCES
+                ${ICEBERG_IO_SOURCES}
+                PRIVATE_INCLUDES
+                ${ICEBERG_IO_INCLUDES})
+
+iceberg_install_all_headers(iceberg/io)

src/iceberg/io/fs_file_io.cc

@@ -0,0 +1,90 @@
+/*
+ * 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/io/fs_file_io.h"
+
+#include <fcntl.h>
+
+#include <cassert>
+#include <filesystem>
+#include <format>
+
+#include <sys/stat.h>
+
+#include "iceberg/exception.h"
+
+namespace iceberg::io {
+
+bool FsInputFile::exists() const { return std::filesystem::exists(location()); }
+
+int64_t FsInputFile::getLength() const {
+  struct stat stat_buffer;
+  if (stat(location().c_str(), &stat_buffer) != 0) {
+    throw IcebergError(std::format(
+        "Failed to get file length. File does not exist or is inaccessible: {}",
+        location_));
+  }
+  return stat_buffer.st_size;
+}
+
+void FsOutputFile::create() {
+  // Check if the file already exists
+  std::ifstream existing_file(location_);
+  bool file_exists = existing_file.good();
+  existing_file.close();
+
+  if (file_exists) {
+    throw IcebergError(std::format("File already exists: {}", location_));
+  }
+
+  // Create or overwrite the file by opening it in truncating mode
+  std::ofstream new_file(location_, std::ios::binary | std::ios::out | std::ios::trunc);
+  if (!new_file.is_open()) {
+    throw IcebergError(std::format("Failed to create or overwrite file: {}", location_));
+  }
+  new_file.close();
+}
+
+std::shared_ptr<InputFile> FsFileIO::newInputFile(const std::string& location) {
+  // Check if the file exists
+  if (!fileExists(location)) {
+    throw IcebergError(std::format("InputFile does not exist: {}", location));
+  }
+
+  // Create and return an FsInputFile instance
+  return std::make_shared<FsInputFile>(location);
+}
+
+std::shared_ptr<OutputFile> FsFileIO::newOutputFile(const std::string& location) {
+  return std::make_shared<FsOutputFile>(location);
+}
+
+void FsFileIO::DeleteFile(const std::string& location) {
+  // Check if the file exists
+  if (!fileExists(location)) {
+    throw IcebergError(std::format("InputFile does not exist: {}", location));
+  }
+  std::error_code ec;
+  if (std::filesystem::remove(location, ec) == false) {
+    throw IcebergError(
+        std::format("Failed to delete file: {}, error code: {}", location, ec.message()));
+  }
+}
+
+}  // namespace iceberg::io

src/iceberg/io/fs_file_io.h

@@ -0,0 +1,102 @@
+/*
+ * 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 "iceberg/file_io.h"
+#include "iceberg/io/fs_file_reader.h"
+#include "iceberg/io/fs_file_writer.h"
+
+namespace iceberg::io {
+
+class ICEBERG_EXPORT FsInputFile : public InputFile {
+ public:
+  explicit FsInputFile(std::string location) : InputFile(std::move(location)) {}
+  ~FsInputFile() = default;
+
+  /// \brief Checks whether the file exists.
+  bool exists() const override;
+
+  /// \brief Returns the total length of the file, in bytes.
+  int64_t getLength() const override;
+
+  /// \brief Get a Reader instance to read bytes from the file.
+  std::unique_ptr<Reader> newReader() override {
+    return std::make_unique<FsFileReader>(location_);
+  }
+};
+
+class ICEBERG_EXPORT FsOutputFile : public OutputFile {
+ public:
+  explicit FsOutputFile(std::string location) : OutputFile(std::move(location)) {}
+  ~FsOutputFile() = default;
+
+  /// \brief Create the file, optionally overwriting if it exists.
+  ///
+  /// If the file already exists, an exception is thrown.
+  void create() override;
+
+  /// \brief Get a Writer instance to write bytes to the file.
+  ///
+  /// Returns a unique pointer to a `FsFileWriter`.
+  std::unique_ptr<Writer> newWriter() override {
+    return std::make_unique<FsFileWriter>(location_);
+  }
+
+  /// \brief Return an InputFile for the location of this OutputFile.
+  ///
+  /// Creates an FsInputFile for reading the file pointed by this OutputFile's location.
+  std::shared_ptr<InputFile> toInputFile() const override {
+    return std::make_shared<FsInputFile>(location_);
+  }
+};
+
+/// \brief A concrete implementation of FileIO for file system.
+class ICEBERG_EXPORT FsFileIO : public FileIO {
+ public:
+  explicit FsFileIO(const std::string& name) : FileIO(name) {}
+
+  ~FsFileIO() override = default;
+
+  /// \brief Get an InputFile
+  ///
+  /// Returns a shared pointer to an FsInputFile instance, representing the file at
+  /// `location`.
+  std::shared_ptr<InputFile> newInputFile(const std::string& location) override;
+
+  /// \brief Get an OutputFile
+  ///
+  /// Returns a shared pointer to an FsOutputFile instance, representing the file at
+  /// `location`.
+  std::shared_ptr<OutputFile> newOutputFile(const std::string& location) override;
+
+  /// \brief Delete file
+  ///
+  /// Deletes the file at the given location using file system operations.
+  void DeleteFile(const std::string& location) override;
+
+ private:
+  /// \brief Check if a file exists
+  bool fileExists(const std::string& location) {
+    std::ifstream file(location);
+    return file.good();
+  }
+};
+
+}  // namespace iceberg::io