refactor(io): make FileIO resolution registry-driven

mkdocs/docs/file-io.md

@@ -0,0 +1,94 @@
+<!--
+  ~ 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.
+-->
+
+# FileIO
+
+`FileIO` reads, writes, and deletes Iceberg data and metadata files.
+
+## Built-in implementations
+
+Call `iceberg::arrow::RegisterAll()` to register the Arrow-backed FileIO
+implementations:
+
+| Registry name | Schemes |
+|---|---|
+| `arrow-fs-local` | paths without a scheme, `file` |
+| `arrow-fs-s3` | `s3`, `s3a`, `s3n` |
+
+The S3 implementation requires Arrow S3 support.
+
+## Select an implementation
+
+Load a registered implementation directly:
+
+```cpp
+#include "iceberg/arrow/arrow_register.h"
+#include "iceberg/arrow/s3/s3_properties.h"
+#include "iceberg/file_io_registry.h"
+#include "iceberg/resolving_file_io.h"
+
+iceberg::arrow::RegisterAll();
+
+auto file_io = iceberg::FileIORegistry::Load(
+    iceberg::FileIORegistry::kArrowS3FileIO,
+    {{std::string(iceberg::arrow::S3Properties::kEndpoint),
+      "https://s3.example.com"},
+     {std::string(iceberg::arrow::S3Properties::kClientRegion), "us-east-1"}});
+```
+
+For a REST catalog, set `io-impl` to the registry name. If it is omitted, the
+REST catalog uses `ResolvingFileIO` and selects a registered implementation for
+each file location's scheme.
+
+## Register a custom FileIO
+
+Register the factory before creating the catalog or resolver:
+
+```cpp
+#include <string_view>
+
+iceberg::FileIORegistry::Register(
+    "my-file-io",
+    {.create = [](const iceberg::FileIORegistry::Properties& properties)
+                   -> iceberg::Result<std::unique_ptr<iceberg::FileIO>> {
+       return MakeMyFileIO(properties);
+     },
+     .accepts = [](std::string_view scheme) { return scheme == "myfs"; }});
+```
+
+`create` is required. Set `accepts` to enable automatic selection; it receives
+the normalized lower-case scheme. Leave it empty for an implementation selected
+only by `io-impl`.
+
+```cpp
+iceberg::FileIORegistry::Load("my-file-io", {});
+auto file_io = std::make_unique<iceberg::ResolvingFileIO>(
+    iceberg::FileIORegistry::Properties{});
+file_io->NewInputFile("myfs://bucket/path/file.parquet");
+```
+
+Later registrations take precedence for automatic scheme resolution.
+
+## Storage credentials
+
+When a REST catalog returns vended storage credentials for a table, it applies
+them to the table's FileIO. A custom FileIO selected through `io-impl` must
+implement `SupportsStorageCredentials`; otherwise table access with vended
+credentials is unsupported. With automatic resolution, `ResolvingFileIO`
+forwards credentials to registered delegates that support them.

mkdocs/mkdocs.yml

@@ -52,6 +52,7 @@ markdown_extensions:
 nav:
   - Home: index.md
   - Getting Started: getting-started.md
+  - FileIO: file-io.md
   - Contributing: contributing.md
   - Releases:
     - Release History: releases.md

src/iceberg/CMakeLists.txt

@@ -135,6 +135,7 @@ set(ICEBERG_SOURCES
     util/conversions.cc
     util/decimal.cc
     util/gzip_internal.cc
+    util/location_util.cc
     util/murmurhash3_internal.cc
     util/property_util.cc
     util/retry_util.cc

src/iceberg/arrow/arrow_register.cc

@@ -21,9 +21,10 @@
 
 #include <mutex>
 #include <string>
-#include <unordered_map>
+#include <string_view>
 
 #include "iceberg/arrow/arrow_io_util.h"
+#include "iceberg/arrow/s3/s3_properties.h"
 #include "iceberg/file_io_registry.h"
 
 namespace iceberg {
@@ -37,16 +38,19 @@ namespace {
 void RegisterLocalFileIO() {
   FileIORegistry::Register(
       std::string(FileIORegistry::kArrowLocalFileIO),
-      [](const std::unordered_map<std::string, std::string>& /*properties*/)
-          -> Result<std::unique_ptr<FileIO>> { return MakeLocalFileIO(); });
+      {.create = [](const FileIORegistry::Properties& /*properties*/)
+           -> Result<std::unique_ptr<FileIO>> { return MakeLocalFileIO(); },
+       .accepts =
+           [](std::string_view scheme) { return scheme.empty() || scheme == "file"; }});
 }
 
 void RegisterS3FileIO() {
 #if ICEBERG_S3_ENABLED
   FileIORegistry::Register(
       std::string(FileIORegistry::kArrowS3FileIO),
-      [](const std::unordered_map<std::string, std::string>& properties)
-          -> Result<std::unique_ptr<FileIO>> { return MakeS3FileIO(properties); });
+      {.create = [](const FileIORegistry::Properties& properties)
+           -> Result<std::unique_ptr<FileIO>> { return MakeS3FileIO(properties); },
+       .accepts = [](std::string_view scheme) { return IsS3Scheme(scheme); }});
 #endif
 }
 

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -91,11 +91,10 @@ std::string SplitEndpointScheme(std::string_view endpoint,
   return std::string(endpoint);
 }
 
-// Location prefixes this FileIO can serve: must cover every scheme
-// ResolveFileIOName routes here, or such a credential would be dropped.
+// Location prefixes this FileIO can serve.
 bool IsS3FileIOCredentialPrefix(std::string_view prefix) {
   return prefix == "s3" || prefix.starts_with("s3://") || prefix.starts_with("s3a://") ||
-         prefix.starts_with("s3n://") || prefix.starts_with("oss://");
+         prefix.starts_with("s3n://");
 }
 
 }  // namespace
@@ -184,9 +183,8 @@ Result<std::shared_ptr<::arrow::fs::FileSystem>> BuildArrowS3FileSystem(
   return std::shared_ptr<::arrow::fs::FileSystem>(std::move(fs));
 }
 
-// Keep in sync with ResolveFileIOName (resolving_file_io.cc).
 std::string CanonicalizeS3Scheme(std::string_view location) {
-  for (std::string_view scheme : {"s3a://", "s3n://", "oss://"}) {
+  for (std::string_view scheme : {"s3a://", "s3n://"}) {
     if (location.starts_with(scheme)) {
       return std::string("s3://").append(location.substr(scheme.size()));
     }

src/iceberg/arrow/s3/s3_properties.h

@@ -22,6 +22,8 @@
 /// \file iceberg/arrow/s3/s3_properties.h
 /// \brief Define S3 configuration property keys.
 
+#include <algorithm>
+#include <array>
 #include <string_view>
 
 namespace iceberg::arrow {
@@ -54,4 +56,15 @@ struct S3Properties {
   static constexpr std::string_view kSocketTimeoutMs = "s3.socket-timeout-ms";
 };
 
+/// \brief URI schemes served by the Arrow S3 FileIO, lower-case.
+///
+/// Single source of truth: both the registry registration and IsS3Scheme derive
+/// from this list, so a new alias only has to be added here.
+inline constexpr std::array<std::string_view, 3> kS3Schemes = {"s3", "s3a", "s3n"};
+
+/// \brief Return whether a normalized URI scheme is S3-compatible.
+inline constexpr bool IsS3Scheme(std::string_view scheme) {
+  return std::ranges::contains(kS3Schemes, scheme);
+}
+
 }  // namespace iceberg::arrow

src/iceberg/catalog/rest/rest_file_io.cc

@@ -26,6 +26,7 @@
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
+#include "iceberg/resolving_file_io.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -45,16 +46,11 @@ std::unordered_map<std::string, std::string> MergeFileIOProperties(
 }  // namespace
 
 Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& config) {
-  std::string io_impl = config.Get(RestCatalogProperties::kIOImpl);
+  const std::string io_impl = config.Get(RestCatalogProperties::kIOImpl);
   if (io_impl.empty()) {
-    // 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);
+    return std::make_unique<ResolvingFileIO>(config.configs());
   }
 
-  // TODO(gangwu): Support Java-style customized FileIO creation flows instead of
-  // resolving a single catalog-scoped FileIO instance only from properties.
   return FileIORegistry::Load(io_impl, config.configs());
 }
 

src/iceberg/file_io_registry.cc

@@ -19,56 +19,98 @@
 
 #include "iceberg/file_io_registry.h"
 
+#include <algorithm>
 #include <mutex>
+#include <ranges>
+#include <string>
 #include <utility>
+#include <vector>
 
-#include "iceberg/resolving_file_io.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
 
 namespace iceberg {
 
 namespace {
 
 struct RegistryState {
-  std::mutex mutex;
-  std::unordered_map<std::string, FileIORegistry::Factory> registry;
+  struct Entry {
+    std::string name;
+    FileIORegistry::Factory factory;
+  };
 
-  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);
-    };
-  }
+  std::mutex mutex;
+  std::vector<Entry> registrations;
 };
 
 RegistryState& State() {
   static RegistryState state;
   return state;
 }
 
+// Copy entries so user callbacks run outside the registry lock.
+std::vector<RegistryState::Entry> SnapshotEntries() {
+  auto& state = State();
+  std::lock_guard lock(state.mutex);
+  return state.registrations;
+}
+
+std::string FormatNames(const std::vector<RegistryState::Entry>& entries) {
+  std::string result;
+  for (const auto& entry : entries) {
+    if (!result.empty()) {
+      result += ", ";
+    }
+    result += entry.name;
+  }
+  return result.empty() ? "(none registered)" : result;
+}
+
 }  // namespace
 
-void FileIORegistry::Register(const std::string& name, Factory factory) {
+void FileIORegistry::Register(std::string name, Factory factory) {
   auto& state = State();
   std::lock_guard lock(state.mutex);
-  state.registry[name] = std::move(factory);
+  std::erase_if(state.registrations, [&name](const RegistryState::Entry& entry) {
+    return entry.name == name;
+  });
+  state.registrations.emplace_back(std::move(name), std::move(factory));
 }
 
-Result<std::unique_ptr<FileIO>> FileIORegistry::Load(
-    const std::string& name,
-    const std::unordered_map<std::string, std::string>& properties) {
-  Factory factory;
+Result<std::unique_ptr<FileIO>> FileIORegistry::Load(std::string_view name,
+                                                     const Properties& properties) {
+  Factory::CreateFunction create;
   {
     auto& state = State();
     std::lock_guard lock(state.mutex);
-    auto it = state.registry.find(name);
-    if (it == state.registry.end()) {
-      return NotFound("FileIO implementation not found: {}", name);
+    const auto entry =
+        std::ranges::find(state.registrations, name, &RegistryState::Entry::name);
+    if (entry == state.registrations.end()) {
+      return NotFound("FileIO not found: {}", name);
+    }
+    create = entry->factory.create;
+  }
+  if (!create) {
+    return InvalidArgument("FileIO '{}' has no create function", name);
+  }
+  ICEBERG_ASSIGN_OR_RAISE(auto io, create(properties));
+  if (!io) {
+    return InvalidArgument("FileIO '{}' returned a null instance", name);
+  }
+  return std::move(io);
+}
+
+Result<std::string> FileIORegistry::Resolve(std::string_view scheme) {
+  const std::string normalized_scheme = StringUtils::ToLower(scheme);
+  const auto entries = SnapshotEntries();
+  // Newest registrations take precedence.
+  for (const auto& entry : std::ranges::reverse_view(entries)) {
+    if (entry.factory.accepts && entry.factory.accepts(normalized_scheme)) {
+      return entry.name;
     }
-    factory = it->second;
   }
-  return factory(properties);
+  return NotSupported("No FileIO registered for URI scheme '{}'; registered: {}",
+                      normalized_scheme, FormatNames(entries));
 }
 
 }  // namespace iceberg

src/iceberg/file_io_registry.h

@@ -43,27 +43,39 @@ 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>>(
-      const std::unordered_map<std::string, std::string>& properties)>;
+  using Properties = std::unordered_map<std::string, std::string>;
+
+  /// Factory for explicit loading and optional scheme-based routing.
+  struct Factory {
+    using CreateFunction =
+        std::function<Result<std::unique_ptr<FileIO>>(const Properties& properties)>;
+    using AcceptsFunction = std::function<bool(std::string_view scheme)>;
+
+    /// Required for explicit loading.
+    CreateFunction create;
+    /// Receives a lower-case scheme. Empty means explicit-only.
+    AcceptsFunction accepts;
+  };
 
   /// \brief Register a FileIO factory under the given name.
   ///
   /// \param name The implementation name (e.g., "local", "s3")
   /// \param factory The factory function that creates the FileIO instance.
-  static void Register(const std::string& name, Factory factory);
+  static void Register(std::string name, Factory factory);
 
   /// \brief Load a FileIO implementation by name.
   ///
   /// \param name The implementation name to look up.
   /// \param properties Configuration properties to pass to the factory.
   /// \return A unique_ptr to the FileIO instance, or an error if not found.
-  static Result<std::unique_ptr<FileIO>> Load(
-      const std::string& name,
-      const std::unordered_map<std::string, std::string>& properties);
+  static Result<std::unique_ptr<FileIO>> Load(std::string_view name,
+                                              const Properties& properties);
+
+  /// \brief Returns the latest registered factory accepting `scheme`.
+  ///
+  /// Matching is case-insensitive; later registrations take precedence.
+  static Result<std::string> Resolve(std::string_view scheme);
 };
 
 }  // namespace iceberg