Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/iceberg/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ set(ICEBERG_SOURCES
partition_field.cc
partition_spec.cc
partition_summary.cc
resolving_file_io.cc
row/arrow_array_wrapper.cc
row/manifest_wrapper.cc
row/partition_values.cc
Expand Down
20 changes: 17 additions & 3 deletions src/iceberg/arrow/s3/arrow_s3_file_io.cc
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
#include "iceberg/arrow/arrow_io_util.h"
#include "iceberg/arrow/arrow_status_internal.h"
#include "iceberg/arrow/s3/s3_properties.h"
#include "iceberg/logging/logger.h"
#include "iceberg/util/macros.h"
#include "iceberg/util/string_util.h"

Expand Down Expand Up @@ -90,6 +91,9 @@ std::string SplitEndpointScheme(std::string_view endpoint,
return std::string(endpoint);
}

// Deliberately narrower than the schemes this FileIO serves: `oss`-prefixed
// credentials target native OSS connectors, while S3-compatible access is
// vended under an `s3` prefix (servers vend both side by side).
bool IsS3FileIOCredentialPrefix(std::string_view prefix) {
return prefix == "s3" || prefix.starts_with("s3://") || prefix.starts_with("s3a://") ||
prefix.starts_with("s3n://");
Expand Down Expand Up @@ -181,6 +185,7 @@ Result<std::shared_ptr<::arrow::fs::FileSystem>> BuildArrowS3FileSystem(
return std::shared_ptr<::arrow::fs::FileSystem>(std::move(fs));
}

// Keep in sync with ResolvingFileIO::ResolveFileIOName's S3-compatible schemes.
std::string CanonicalizeS3Scheme(std::string_view location) {
for (std::string_view scheme : {"s3a://", "s3n://", "oss://"}) {
if (location.starts_with(scheme)) {
Expand Down Expand Up @@ -235,10 +240,11 @@ Status ArrowS3FileIO::SetStorageCredentials(
// TODO(gangwu): Refresh vended credentials via credentials.uri before tokens expire.
for (const auto& credential : storage_credentials) {
ICEBERG_RETURN_UNEXPECTED(credential.Validate());
// A server may vend credentials for several storage systems at once;
// non-S3 prefixes are skipped, not rejected (Java S3FileIO filters
// credentials by the "s3" prefix).
if (!IsS3FileIOCredentialPrefix(credential.prefix)) {
return NotSupported(
"Storage credential prefix '{}' is unsupported by Arrow S3 FileIO",
credential.prefix);
continue;
}
auto properties = default_properties_;
for (const auto& [key, value] : credential.config) {
Expand All @@ -249,6 +255,14 @@ Status ArrowS3FileIO::SetStorageCredentials(
CanonicalizeS3Scheme(credential.prefix),
std::make_unique<ArrowFileSystemFileIO>(std::move(fs)));
}
if (file_io_by_prefix.empty() && !storage_credentials.empty()) {
// Silent skipping of every vended credential is hard to diagnose: S3 access
// would proceed with the default credentials and fail only at IO time.
Log(LogLevel::kWarn,
"None of the {} vended storage credential(s) has an S3-compatible prefix; "
"S3 access will use the default credentials",
storage_credentials.size());
}
file_io_by_prefix_ = std::move(file_io_by_prefix);
storage_credentials_ = storage_credentials;
return {};
Expand Down
72 changes: 6 additions & 66 deletions src/iceberg/catalog/rest/rest_file_io.cc
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@

#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "iceberg/catalog/rest/types.h"
Expand All @@ -33,11 +32,6 @@ namespace iceberg::rest {

namespace {

bool IsBuiltinImpl(std::string_view io_impl) {
return io_impl == FileIORegistry::kArrowLocalFileIO ||
io_impl == FileIORegistry::kArrowS3FileIO;
}

std::unordered_map<std::string, std::string> MergeFileIOProperties(
const std::unordered_map<std::string, std::string>& catalog_config,
const std::unordered_map<std::string, std::string>& table_config) {
Expand All @@ -50,56 +44,13 @@ std::unordered_map<std::string, std::string> MergeFileIOProperties(

} // namespace

Result<BuiltinFileIOKind> DetectBuiltinFileIO(std::string_view location) {
const auto pos = location.find("://");
if (pos == std::string_view::npos) {
return BuiltinFileIOKind::kArrowLocal;
}

const auto scheme = location.substr(0, pos);
if (scheme == "file") {
return BuiltinFileIOKind::kArrowLocal;
}
if (scheme == "s3" || scheme == "s3a" || scheme == "s3n") {
return BuiltinFileIOKind::kArrowS3;
}

return NotSupported("URI scheme '{}' is not supported for automatic FileIO resolution",
scheme);
}

std::string_view BuiltinFileIOName(BuiltinFileIOKind kind) {
switch (kind) {
case BuiltinFileIOKind::kArrowLocal:
return FileIORegistry::kArrowLocalFileIO;
case BuiltinFileIOKind::kArrowS3:
return FileIORegistry::kArrowS3FileIO;
}
std::unreachable();
}

Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& config) {
std::string io_impl = config.Get(RestCatalogProperties::kIOImpl);
std::string warehouse = config.Get(RestCatalogProperties::kWarehouse);

if (io_impl.empty()) {
if (warehouse.empty()) {
return InvalidArgument(R"("{}" or "{}" property is required to create FileIO)",
RestCatalogProperties::kIOImpl.key(),
RestCatalogProperties::kWarehouse.key());
}
ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
io_impl = std::string(BuiltinFileIOName(detected_kind));
}

if (!warehouse.empty() && IsBuiltinImpl(io_impl)) {
ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
const auto detected_name = BuiltinFileIOName(detected_kind);
if (io_impl != detected_name) {
return InvalidArgument(
R"("io-impl" value '{}' is incompatible with warehouse '{}')", io_impl,
warehouse);
}
// Resolve the FileIO per file-path scheme instead of guessing from
// `warehouse`, which is often a logical identifier rather than a storage
// URI (Java defaults to ResolvingFileIO likewise).
io_impl = std::string(FileIORegistry::kResolvingFileIO);
}

// TODO(gangwu): Support Java-style customized FileIO creation flows instead of
Expand All @@ -112,19 +63,8 @@ Result<std::unique_ptr<FileIO>> MakeTableFileIO(
const std::unordered_map<std::string, std::string>& table_config,
const std::vector<StorageCredential>& storage_credentials) {
const auto default_properties = MergeFileIOProperties(catalog_config, table_config);
const auto properties = RestCatalogProperties::FromMap(default_properties);
auto io_impl = properties.Get(RestCatalogProperties::kIOImpl);
if (io_impl.empty()) {
const auto warehouse = properties.Get(RestCatalogProperties::kWarehouse);
if (warehouse.empty()) {
return InvalidArgument(R"("{}" or "{}" property is required to create FileIO)",
RestCatalogProperties::kIOImpl.key(),
RestCatalogProperties::kWarehouse.key());
}
ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
io_impl = std::string(BuiltinFileIOName(detected_kind));
}
ICEBERG_ASSIGN_OR_RAISE(auto io, FileIORegistry::Load(io_impl, default_properties));
ICEBERG_ASSIGN_OR_RAISE(
auto io, MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties)));

if (storage_credentials.empty()) {
return io;
Expand Down
14 changes: 2 additions & 12 deletions src/iceberg/catalog/rest/rest_file_io.h
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,7 @@
/// \file iceberg/catalog/rest/rest_file_io.h
/// \brief Provide helpers to create FileIO instances for REST catalog responses.

#include <cstdint>
#include <memory>
#include <string_view>
#include <unordered_map>
#include <vector>

Expand All @@ -37,16 +35,8 @@

namespace iceberg::rest {

enum class BuiltinFileIOKind : uint8_t {
kArrowLocal,
kArrowS3,
};

ICEBERG_REST_EXPORT Result<BuiltinFileIOKind> DetectBuiltinFileIO(
std::string_view location);

ICEBERG_REST_EXPORT std::string_view BuiltinFileIOName(BuiltinFileIOKind kind);

/// \brief Build the catalog FileIO: the configured `io-impl`, or the
/// scheme-resolving FileIO by default.
ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
const RestCatalogProperties& config);

Expand Down
11 changes: 11 additions & 0 deletions src/iceberg/file_io_registry.cc
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,24 @@
#include <mutex>
#include <utility>

#include "iceberg/resolving_file_io.h"

namespace iceberg {

namespace {

struct RegistryState {
std::mutex mutex;
std::unordered_map<std::string, FileIORegistry::Factory> registry;

RegistryState() {
// Always available: the scheme-resolving FileIO lives in the core library.
registry[std::string(FileIORegistry::kResolvingFileIO)] =
[](const std::unordered_map<std::string, std::string>& properties)
-> Result<std::unique_ptr<FileIO>> {
return std::make_unique<ResolvingFileIO>(properties);
};
}
};

RegistryState& State() {
Expand Down
2 changes: 2 additions & 0 deletions src/iceberg/file_io_registry.h
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,8 @@ class ICEBERG_EXPORT FileIORegistry {
public:
static constexpr std::string_view kArrowLocalFileIO = "arrow-fs-local";
static constexpr std::string_view kArrowS3FileIO = "arrow-fs-s3";
/// Always registered; resolves the concrete FileIO per file-path scheme.
static constexpr std::string_view kResolvingFileIO = "resolving-file-io";

/// Factory function type for creating FileIO instances.
using Factory = std::function<Result<std::unique_ptr<FileIO>>(
Expand Down
2 changes: 2 additions & 0 deletions src/iceberg/meson.build
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@ iceberg_sources = files(
'partition_field.cc',
'partition_spec.cc',
'partition_summary.cc',
'resolving_file_io.cc',
'row/arrow_array_wrapper.cc',
'row/manifest_wrapper.cc',
'row/partition_values.cc',
Expand Down Expand Up @@ -295,6 +296,7 @@ install_headers(
'name_mapping.h',
'partition_field.h',
'partition_spec.h',
'resolving_file_io.h',
'result.h',
'schema_field.h',
'schema.h',
Expand Down
126 changes: 126 additions & 0 deletions src/iceberg/resolving_file_io.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
/*
* 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/resolving_file_io.h"

#include <utility>

#include "iceberg/file_io_registry.h"
#include "iceberg/util/macros.h"

namespace iceberg {

ResolvingFileIO::ResolvingFileIO(std::unordered_map<std::string, std::string> properties)
: properties_(std::move(properties)) {}

ResolvingFileIO::~ResolvingFileIO() = default;

Result<std::string_view> ResolvingFileIO::ResolveFileIOName(std::string_view location) {
const auto pos = location.find("://");
if (pos == std::string_view::npos) {
return FileIORegistry::kArrowLocalFileIO;
}

const auto scheme = location.substr(0, pos);
if (scheme == "file") {
return FileIORegistry::kArrowLocalFileIO;
}
// S3-compatible schemes served by the S3 FileIO (Java: SCHEME_TO_FILE_IO).
// Keep in sync with CanonicalizeS3Scheme in arrow_s3_file_io.cc.
if (scheme == "s3" || scheme == "s3a" || scheme == "s3n" || scheme == "oss") {
return FileIORegistry::kArrowS3FileIO;
}

return NotSupported("URI scheme '{}' is not supported for FileIO resolution", scheme);
}

Result<FileIO*> ResolvingFileIO::FileIOForPath(std::string_view location) {
ICEBERG_ASSIGN_OR_RAISE(const auto name, ResolveFileIOName(location));

std::lock_guard lock(mutex_);
auto it = io_by_name_.find(name);
if (it == io_by_name_.end()) {
ICEBERG_ASSIGN_OR_RAISE(auto io,
FileIORegistry::Load(std::string(name), properties_));
// Forward all credentials; each implementation applies the prefixes it
// understands.
if (!storage_credentials_.empty()) {
if (auto* credentialed = io->AsSupportsStorageCredentials()) {
ICEBERG_RETURN_UNEXPECTED(
credentialed->SetStorageCredentials(storage_credentials_));
}
}
it = io_by_name_.emplace(std::string(name), std::move(io)).first;
}
return it->second.get();
}

Result<std::unique_ptr<InputFile>> ResolvingFileIO::NewInputFile(
std::string file_location) {
ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location));
return io->NewInputFile(std::move(file_location));
}

Result<std::unique_ptr<InputFile>> ResolvingFileIO::NewInputFile(
std::string file_location, size_t length) {
ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location));
return io->NewInputFile(std::move(file_location), length);
}

Result<std::unique_ptr<OutputFile>> ResolvingFileIO::NewOutputFile(
std::string file_location) {
ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location));
return io->NewOutputFile(std::move(file_location));
}

Status ResolvingFileIO::DeleteFile(const std::string& file_location) {
ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location));
return io->DeleteFile(file_location);
}

Status ResolvingFileIO::DeleteFiles(const std::vector<std::string>& file_locations) {
std::unordered_map<FileIO*, std::vector<std::string>> locations_by_io;
for (const auto& file_location : file_locations) {
ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location));
locations_by_io[io].push_back(file_location);
}
for (auto& [io, locations] : locations_by_io) {
ICEBERG_RETURN_UNEXPECTED(io->DeleteFiles(locations));
}
return {};
}

Status ResolvingFileIO::SetStorageCredentials(
const std::vector<StorageCredential>& storage_credentials) {
std::lock_guard lock(mutex_);
storage_credentials_ = storage_credentials;
for (auto& [name, io] : io_by_name_) {
if (auto* credentialed = io->AsSupportsStorageCredentials()) {
ICEBERG_RETURN_UNEXPECTED(
credentialed->SetStorageCredentials(storage_credentials_));
}
}
return {};
}

const std::vector<StorageCredential>& ResolvingFileIO::credentials() const {
return storage_credentials_;
}

} // namespace iceberg
Loading
Loading