feat(io): add ResolvingFileIO to resolve FileIO by location scheme and forward vended credentials

src/iceberg/CMakeLists.txt

@@ -79,6 +79,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

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -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/log_macros.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/string_util.h"
 
@@ -90,9 +91,11 @@ 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.
 bool IsS3FileIOCredentialPrefix(std::string_view prefix) {
   return prefix == "s3" || prefix.starts_with("s3://") || prefix.starts_with("s3a://") ||
-         prefix.starts_with("s3n://");
+         prefix.starts_with("s3n://") || prefix.starts_with("oss://");
 }
 
 }  // namespace
@@ -181,6 +184,7 @@ 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://"}) {
     if (location.starts_with(scheme)) {
@@ -235,10 +239,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) {
@@ -249,6 +254,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.
+    ICEBERG_LOG_WARN(
+        "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 {};

src/iceberg/catalog/rest/rest_file_io.cc

@@ -21,7 +21,6 @@
 
 #include <string>
 #include <unordered_map>
-#include <utility>
 #include <vector>
 
 #include "iceberg/catalog/rest/types.h"
@@ -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) {
@@ -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
@@ -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;

src/iceberg/catalog/rest/rest_file_io.h

@@ -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>
 
@@ -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);
 

src/iceberg/file_io_registry.cc

@@ -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() {

src/iceberg/file_io_registry.h

@@ -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>>(

src/iceberg/meson.build

@@ -128,6 +128,7 @@ iceberg_sources = files(
     'partition_spec.cc',
     'partition_summary.cc',
     'puffin_dv_io.cc',
+    'resolving_file_io.cc',
     'row/arrow_array_wrapper.cc',
     'row/manifest_wrapper.cc',
     'row/partition_values.cc',
@@ -324,6 +325,7 @@ install_headers(
         'name_mapping.h',
         'partition_field.h',
         'partition_spec.h',
+        'resolving_file_io.h',
         'result.h',
         'schema_field.h',
         'schema.h',

src/iceberg/resolving_file_io.cc

@@ -0,0 +1,124 @@
+/*
+ * 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/resolving_file_io_internal.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> 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) {
+  // Rebuild delegates lazily with the new credentials. Updating live delegates
+  // instead would leave the resolver inconsistent if one of them rejected them.
+  std::lock_guard lock(mutex_);
+  storage_credentials_ = storage_credentials;
+  io_by_name_.clear();
+  return {};
+}
+
+const std::vector<StorageCredential>& ResolvingFileIO::credentials() const {
+  return storage_credentials_;
+}
+
+}  // namespace iceberg