| /* | |
| * 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 <cstddef> | |
| #include <cstdint> | |
| #include <filesystem> | |
| #include <fstream> | |
| #include <ios> | |
| #include <limits> | |
| #include <memory> | |
| #include <optional> | |
| #include <span> | |
| #include <string> | |
| #include <string_view> | |
| #include <system_error> | |
| #include <utility> | |
| #include "iceberg/file_io.h" | |
| #include "iceberg/result.h" | |
| #include "iceberg/util/macros.h" | |
| namespace iceberg::test { | |
| namespace detail { | |
| inline Result<std::streamsize> ToStreamSize(size_t size) { | |
| if (size > static_cast<size_t>(std::numeric_limits<std::streamsize>::max())) { | |
| return InvalidArgument("Buffer size {} exceeds streamsize max", size); | |
| } | |
| return static_cast<std::streamsize>(size); | |
| } | |
| inline Result<int64_t> ToInt64FileSize(uintmax_t size, std::string_view location) { | |
| if (size > static_cast<uintmax_t>(std::numeric_limits<int64_t>::max())) { | |
| return Invalid("File size for {} exceeds int64_t max", location); | |
| } | |
| return static_cast<int64_t>(size); | |
| } | |
| } // namespace detail | |
| class StdSeekableInputStream : public SeekableInputStream { | |
| public: | |
| explicit StdSeekableInputStream(std::string location) | |
| : location_(std::move(location)), file_(location_, std::ios::binary) {} | |
| bool is_open() const { return file_.is_open(); } | |
| Result<int64_t> Position() const override { | |
| auto position = file_.tellg(); | |
| if (position < 0) { | |
| return IOError("Failed to get read position for: {}", location_); | |
| } | |
| return static_cast<int64_t>(position); | |
| } | |
| Status Seek(int64_t position) override { | |
| if (position < 0) { | |
| return InvalidArgument("Cannot seek to negative position {}", position); | |
| } | |
| file_.clear(); | |
| file_.seekg(position); | |
| if (!file_) { | |
| return IOError("Failed to seek to {} in file: {}", position, location_); | |
| } | |
| return {}; | |
| } | |
| Result<int64_t> Read(std::span<std::byte> out) override { | |
| if (out.empty()) { | |
| return 0; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto read_size, detail::ToStreamSize(out.size())); | |
| file_.read(reinterpret_cast<char*>(out.data()), read_size); | |
| auto bytes_read = file_.gcount(); | |
| if (!file_) { | |
| if (file_.bad() || !file_.eof()) { | |
| return IOError("Failed to read from file: {}", location_); | |
| } | |
| file_.clear(); | |
| } | |
| if (bytes_read < 0) { | |
| return IOError("Failed to read from file: {}", location_); | |
| } | |
| return static_cast<int64_t>(bytes_read); | |
| } | |
| Status ReadFully(int64_t position, std::span<std::byte> out) override { | |
| if (position < 0) { | |
| return InvalidArgument("Cannot read from negative position {}", position); | |
| } | |
| if (out.empty()) { | |
| return {}; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto original_position, Position()); | |
| auto seek_status = Seek(position); | |
| if (!seek_status.has_value()) { | |
| static_cast<void>(Seek(original_position)); | |
| return seek_status; | |
| } | |
| Status read_status = {}; | |
| size_t total_read = 0; | |
| while (total_read < out.size()) { | |
| auto read_result = Read(out.subspan(total_read)); | |
| if (!read_result.has_value()) { | |
| read_status = ::iceberg::unexpected<Error>(read_result.error()); | |
| break; | |
| } | |
| if (read_result.value() == 0) { | |
| read_status = | |
| IOError("Failed to read {} bytes from file: {}", out.size(), location_); | |
| break; | |
| } | |
| total_read += static_cast<size_t>(read_result.value()); | |
| } | |
| auto restore_status = Seek(original_position); | |
| ICEBERG_RETURN_UNEXPECTED(read_status); | |
| return restore_status; | |
| } | |
| Status Close() override { | |
| if (!file_.is_open()) { | |
| return {}; | |
| } | |
| file_.close(); | |
| if (!file_) { | |
| return IOError("Failed to close file: {}", location_); | |
| } | |
| return {}; | |
| } | |
| private: | |
| std::string location_; | |
| mutable std::ifstream file_; | |
| }; | |
| class StdPositionOutputStream : public PositionOutputStream { | |
| public: | |
| explicit StdPositionOutputStream(std::string location) | |
| : location_(std::move(location)), | |
| file_(location_, std::ios::binary | std::ios::out | std::ios::trunc) {} | |
| bool is_open() const { return file_.is_open(); } | |
| Result<int64_t> Position() const override { | |
| auto position = file_.tellp(); | |
| if (position < 0) { | |
| return IOError("Failed to get write position for: {}", location_); | |
| } | |
| return static_cast<int64_t>(position); | |
| } | |
| Result<int64_t> StoredLength() const override { | |
| if (file_.is_open()) { | |
| return Position(); | |
| } | |
| std::error_code ec; | |
| auto size = std::filesystem::file_size(location_, ec); | |
| if (ec) { | |
| return IOError("Failed to get file size for {}: {}", location_, ec.message()); | |
| } | |
| return detail::ToInt64FileSize(size, location_); | |
| } | |
| Status Write(std::span<const std::byte> data) override { | |
| if (data.empty()) { | |
| return {}; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto write_size, detail::ToStreamSize(data.size())); | |
| file_.write(reinterpret_cast<const char*>(data.data()), write_size); | |
| if (!file_) { | |
| return IOError("Failed to write to file: {}", location_); | |
| } | |
| return {}; | |
| } | |
| Status Flush() override { | |
| file_.flush(); | |
| if (!file_) { | |
| return IOError("Failed to flush file: {}", location_); | |
| } | |
| return {}; | |
| } | |
| Status Close() override { | |
| if (!file_.is_open()) { | |
| return {}; | |
| } | |
| file_.close(); | |
| if (!file_) { | |
| return IOError("Failed to close file: {}", location_); | |
| } | |
| return {}; | |
| } | |
| private: | |
| std::string location_; | |
| mutable std::ofstream file_; | |
| }; | |
| class StdInputFile : public InputFile { | |
| public: | |
| explicit StdInputFile(std::string location, | |
| std::optional<int64_t> file_size = std::nullopt) | |
| : location_(std::move(location)), file_size_(file_size) {} | |
| std::string_view location() const override { return location_; } | |
| Result<int64_t> Size() const override { | |
| if (file_size_.has_value()) { | |
| return *file_size_; | |
| } | |
| std::error_code ec; | |
| auto size = std::filesystem::file_size(location_, ec); | |
| if (ec) { | |
| return IOError("Failed to get file size for {}: {}", location_, ec.message()); | |
| } | |
| return detail::ToInt64FileSize(size, location_); | |
| } | |
| Result<std::unique_ptr<SeekableInputStream>> Open() override { | |
| auto stream = std::make_unique<StdSeekableInputStream>(location_); | |
| if (!stream->is_open()) { | |
| return IOError("Failed to open file for reading: {}", location_); | |
| } | |
| return stream; | |
| } | |
| private: | |
| std::string location_; | |
| std::optional<int64_t> file_size_; | |
| }; | |
| class StdOutputFile : public OutputFile { | |
| public: | |
| explicit StdOutputFile(std::string location) : location_(std::move(location)) {} | |
| std::string_view location() const override { return location_; } | |
| Result<std::unique_ptr<PositionOutputStream>> Create() override { | |
| return Create(/*overwrite=*/false); | |
| } | |
| Result<std::unique_ptr<PositionOutputStream>> CreateOrOverwrite() override { | |
| return Create(/*overwrite=*/true); | |
| } | |
| private: | |
| Result<std::unique_ptr<PositionOutputStream>> Create(bool overwrite) { | |
| std::filesystem::path path(location_); | |
| std::error_code ec; | |
| auto exists = std::filesystem::exists(path, ec); | |
| if (ec) { | |
| return IOError("Failed to check file existence for {}: {}", location_, | |
| ec.message()); | |
| } | |
| if (!overwrite && exists) { | |
| return AlreadyExists("File already exists: {}", location_); | |
| } | |
| if (path.has_parent_path()) { | |
| std::filesystem::create_directories(path.parent_path(), ec); | |
| if (ec) { | |
| return IOError("Failed to create parent directories for {}: {}", location_, | |
| ec.message()); | |
| } | |
| } | |
| auto stream = std::make_unique<StdPositionOutputStream>(location_); | |
| if (!stream->is_open()) { | |
| return IOError("Failed to open file for writing: {}", location_); | |
| } | |
| return stream; | |
| } | |
| std::string location_; | |
| }; | |
| /// \brief Simple local filesystem FileIO implementation for testing | |
| /// | |
| /// This class provides a basic FileIO implementation that reads and writes | |
| /// files to the local filesystem using standard C++ file streams. | |
| class StdFileIO : public FileIO { | |
| public: | |
| Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location) override { | |
| return std::make_unique<StdInputFile>(std::move(file_location)); | |
| } | |
| Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location, | |
| size_t length) override { | |
| if (length > static_cast<size_t>(std::numeric_limits<int64_t>::max())) { | |
| return InvalidArgument("File length {} exceeds int64_t max", length); | |
| } | |
| return std::make_unique<StdInputFile>(std::move(file_location), | |
| static_cast<int64_t>(length)); | |
| } | |
| Result<std::unique_ptr<OutputFile>> NewOutputFile(std::string file_location) override { | |
| return std::make_unique<StdOutputFile>(std::move(file_location)); | |
| } | |
| Status DeleteFile(const std::string& file_location) override { | |
| std::error_code ec; | |
| if (!std::filesystem::remove(file_location, ec)) { | |
| if (ec) { | |
| return IOError("Failed to delete file {}: {}", file_location, ec.message()); | |
| } | |
| return IOError("File does not exist: {}", file_location); | |
| } | |
| return {}; | |
| } | |
| }; | |
| } // namespace iceberg::test |