blob: 725fc7ba5275b7986b1efbfa23d8bc1e940a4f39 [file]
/*
* 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 = std::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