fix(rest): resolve a FileIO without io-impl or warehouse configured

src/iceberg/catalog/rest/rest_catalog.cc

@@ -45,6 +45,7 @@
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/sort_order.h"
+#include "iceberg/storage_credential.h"
 #include "iceberg/table.h"
 #include "iceberg/table_requirement.h"
 #include "iceberg/table_requirements.h"
@@ -488,9 +489,11 @@ Result<std::shared_ptr<auth::AuthSession>> RestCatalog::TableAuthSession(
 Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
     const SessionContext& /*context*/,
     const std::unordered_map<std::string, std::string>& table_config,
+    std::string_view metadata_location,
     const std::vector<StorageCredential>& storage_credentials) const {
   if (!table_config.empty() || !storage_credentials.empty()) {
-    return MakeTableFileIO(config_.configs(), table_config, storage_credentials);
+    return MakeTableFileIO(config_.configs(), table_config, metadata_location,
+                           storage_credentials);
   }
 
   return file_io_;
@@ -730,8 +733,9 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
                           /*stage_create=*/true, *contextual_session));
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto table_io,
+      TableFileIO(context, table_config, result.metadata_location, storage_credentials));
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
@@ -846,8 +850,9 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
     std::shared_ptr<auth::AuthSession> contextual_session) {
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto table_io,
+      TableFileIO(context, table_config, result.metadata_location, storage_credentials));
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));

src/iceberg/catalog/rest/rest_catalog.h

@@ -22,6 +22,7 @@
 #include <memory>
 #include <string>
 #include <unordered_set>
+#include <vector>
 
 #include "iceberg/catalog.h"
 #include "iceberg/catalog/rest/catalog_properties.h"
@@ -81,6 +82,7 @@ class ICEBERG_REST_EXPORT RestCatalog final
   Result<std::shared_ptr<FileIO>> TableFileIO(
       const SessionContext& context,
       const std::unordered_map<std::string, std::string>& table_config,
+      std::string_view metadata_location,
       const std::vector<StorageCredential>& storage_credentials) const;
 
   Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns,

src/iceberg/catalog/rest/rest_file_io.cc

@@ -27,6 +27,7 @@
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
+#include "iceberg/storage_credential.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -84,12 +85,14 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
 
   if (io_impl.empty()) {
     if (warehouse.empty()) {
-      return InvalidArgument(R"("{}" or "{}" property is required to create FileIO)",
-                             RestCatalogProperties::kIOImpl.key(),
-                             RestCatalogProperties::kWarehouse.key());
+      // No io-impl or warehouse configured. Fall back to a local FileIO so the
+      // catalog can still be created; remote access for individual tables is
+      // resolved per table (e.g. from vended storage credentials).
+      io_impl = std::string(FileIORegistry::kArrowLocalFileIO);
+    } else {
+      ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
+      io_impl = std::string(BuiltinFileIOName(detected_kind));
     }
-    ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
-    io_impl = std::string(BuiltinFileIOName(detected_kind));
   }
 
   if (!warehouse.empty() && IsBuiltinImpl(io_impl)) {
@@ -110,19 +113,27 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
 Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
+    std::string_view metadata_location,
     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()) {
+    // With no explicit io-impl, infer it from the table's metadata-location
+    // scheme, then the warehouse scheme, matching PyIceberg's load_file_io. This
+    // lets a catalog that relies solely on vended credentials resolve a table
+    // FileIO without a client-side io-impl or warehouse.
     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());
+    if (!metadata_location.empty()) {
+      ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind,
+                              DetectBuiltinFileIO(metadata_location));
+      io_impl = std::string(BuiltinFileIOName(detected_kind));
+    } else if (!warehouse.empty()) {
+      ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse));
+      io_impl = std::string(BuiltinFileIOName(detected_kind));
+    } else {
+      io_impl = std::string(FileIORegistry::kArrowLocalFileIO);
     }
-    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));
 

src/iceberg/catalog/rest/rest_file_io.h

@@ -34,6 +34,7 @@
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/result.h"
+#include "iceberg/storage_credential.h"
 
 namespace iceberg::rest {
 
@@ -50,10 +51,14 @@ ICEBERG_REST_EXPORT std::string_view BuiltinFileIOName(BuiltinFileIOKind kind);
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
     const RestCatalogProperties& config);
 
-/// \brief Build the configured table FileIO and apply storage credentials if present.
+/// \brief Build the table FileIO and apply storage credentials if present.
+///
+/// When no "io-impl" is configured, the implementation is inferred from the
+/// \p metadata_location scheme, then the "warehouse" scheme.
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
+    std::string_view metadata_location,
     const std::vector<StorageCredential>& storage_credentials);
 
 }  // namespace iceberg::rest

src/iceberg/test/rest_file_io_test.cc

@@ -26,7 +26,7 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
-#include "iceberg/catalog/rest/types.h"
+#include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/test/matchers.h"
 
@@ -88,9 +88,16 @@ TEST(RestFileIOTest, DetectBuiltinKindRejectsUnsupportedScheme) {
   EXPECT_THAT(result, HasErrorMessage("not supported for automatic FileIO resolution"));
 }
 
-TEST(RestFileIOTest, MakeCatalogFileIOMissingImplAndWarehouse) {
+TEST(RestFileIOTest, MakeCatalogFileIOFallsBackToLocalWhenUnconfigured) {
+  FileIORegistry::Register(
+      std::string(FileIORegistry::kArrowLocalFileIO),
+      [](const std::unordered_map<std::string, std::string>& /*properties*/)
+          -> Result<std::unique_ptr<FileIO>> { return std::make_unique<MockFileIO>(); });
+
+  // With neither io-impl nor warehouse configured, the catalog still gets a
+  // (local) FileIO rather than failing, so per-table credential vending works.
   auto result = MakeCatalogFileIO(RestCatalogProperties::default_properties());
-  EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+  ASSERT_THAT(result, IsOk());
 }
 
 TEST(RestFileIOTest, MakeCatalogFileIORejectsIncompatibleWarehouse) {
@@ -187,6 +194,7 @@ TEST(RestFileIOTest, TableFileIOMergesConfigAndCredentials) {
        {"catalog-only", "catalog"},
        {"shared", "catalog"}},
       {{"io-impl", custom_impl}, {"table-only", "table"}, {"shared", "table"}},
+      /*metadata_location=*/"s3://bucket/table/metadata/v1.metadata.json",
       {{.prefix = "s3://bucket/table",
         .config = {{"shared", "credential"}, {"credential-only", "value"}}}});
   ASSERT_THAT(result, IsOk());
@@ -221,7 +229,7 @@ TEST(RestFileIOTest, TableImplOverridesWarehouseScheme) {
   auto result =
       MakeTableFileIO({{"warehouse", "/tmp/catalog-warehouse"}},
                       {{"io-impl", std::string(FileIORegistry::kArrowS3FileIO)}},
-                      /*storage_credentials=*/{});
+                      /*metadata_location=*/"", /*storage_credentials=*/{});
   ASSERT_THAT(result, IsOk());
   EXPECT_THAT(
       captured_file_io_properties,
@@ -239,9 +247,67 @@ TEST(RestFileIOTest, TableFileIORejectsCredentials) {
 
   auto result = MakeTableFileIO(
       {{"warehouse", "s3://catalog/warehouse"}}, {{"io-impl", custom_impl}},
+      /*metadata_location=*/"s3://bucket/table",
       {{.prefix = "s3://bucket/table", .config = {{"k", "v"}}}});
   EXPECT_THAT(result, IsError(ErrorKind::kNotSupported));
   EXPECT_THAT(result, HasErrorMessage("does not support vended storage credentials"));
 }
 
+TEST(RestFileIOTest, TableFileIODetectsImplFromMetadataLocation) {
+  captured_storage_credentials.clear();
+  FileIORegistry::Register(
+      std::string(FileIORegistry::kArrowS3FileIO),
+      [](const std::unordered_map<std::string, std::string>& /*properties*/)
+          -> Result<std::unique_ptr<FileIO>> {
+        return std::make_unique<MockCredentialedFileIO>();
+      });
+
+  // No io-impl or warehouse anywhere; the implementation is inferred from the
+  // table's metadata location scheme. The credential carries a scheme-only "s3"
+  // prefix (which is not itself a resolvable URI), so detection must not rely on
+  // it; the vended credentials are still applied.
+  auto result = MakeTableFileIO(
+      /*catalog_config=*/{}, /*table_config=*/{},
+      /*metadata_location=*/"s3://bucket/table/metadata/v1.metadata.json",
+      {{.prefix = "s3", .config = {{"k", "v"}}}});
+  ASSERT_THAT(result, IsOk());
+  ASSERT_NE(result.value()->AsSupportsStorageCredentials(), nullptr);
+  ASSERT_EQ(captured_storage_credentials.size(), 1);
+  EXPECT_EQ(captured_storage_credentials[0].prefix, "s3");
+}
+
+TEST(RestFileIOTest, TableFileIOPrefersMetadataLocationOverWarehouse) {
+  captured_storage_credentials.clear();
+  FileIORegistry::Register(
+      std::string(FileIORegistry::kArrowS3FileIO),
+      [](const std::unordered_map<std::string, std::string>& /*properties*/)
+          -> Result<std::unique_ptr<FileIO>> {
+        return std::make_unique<MockCredentialedFileIO>();
+      });
+
+  // The warehouse is a logical (non-URI) name that would resolve to a local
+  // FileIO; the s3 metadata location must win so the vended credentials apply.
+  auto result = MakeTableFileIO(
+      /*catalog_config=*/{{"warehouse", "logical-warehouse"}}, /*table_config=*/{},
+      /*metadata_location=*/"s3://bucket/table/metadata/v1.metadata.json",
+      {{.prefix = "s3://bucket/table", .config = {{"k", "v"}}}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_NE(result.value()->AsSupportsStorageCredentials(), nullptr);
+}
+
+TEST(RestFileIOTest, TableFileIOFallsBackToLocalWhenUnconfigured) {
+  FileIORegistry::Register(
+      std::string(FileIORegistry::kArrowLocalFileIO),
+      [](const std::unordered_map<std::string, std::string>& /*properties*/)
+          -> Result<std::unique_ptr<FileIO>> { return std::make_unique<MockFileIO>(); });
+
+  // table_config exercises MakeTableFileIO, but there is no io-impl, warehouse,
+  // or metadata location to infer from, so it defaults to a local FileIO.
+  auto result = MakeTableFileIO(/*catalog_config=*/{},
+                                /*table_config=*/{{"some-key", "value"}},
+                                /*metadata_location=*/"",
+                                /*storage_credentials=*/{});
+  ASSERT_THAT(result, IsOk());
+}
+
 }  // namespace iceberg::rest