diff --git a/src/iceberg/arrow/arrow_io.cc b/src/iceberg/arrow/arrow_io.cc
index 6b159ca89..b409f00c8 100644
--- a/src/iceberg/arrow/arrow_io.cc
+++ b/src/iceberg/arrow/arrow_io.cc
@@ -22,6 +22,7 @@
 #include <limits>
 #include <mutex>
 #include <optional>
+#include <unordered_map>
 #include <vector>
 
 #include <arrow/buffer.h>
@@ -35,6 +36,8 @@
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_io_util.h"
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow/s3/arrow_s3_internal.h"
+#include "iceberg/util/location_util.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::arrow {
@@ -473,13 +476,53 @@ class ArrowOutputFile : public OutputFile {
 
 }  // namespace
 
-Result<std::string> ArrowFileSystemFileIO::ResolvePath(const std::string& file_location) {
+const std::shared_ptr<::arrow::fs::FileSystem>& ArrowFileSystemFileIO::FileSystemForPath(
+    std::string_view location) const {
+  if (fs_by_prefix_.empty()) {
+    return arrow_fs_;
+  }
+  // Longest matching prefix wins; fall back to the default file system.
+  const std::string canonical = LocationUtil::CanonicalizeS3Scheme(location);
+  const std::shared_ptr<::arrow::fs::FileSystem>* best = &arrow_fs_;
+  size_t best_len = 0;
+  for (const auto& [prefix, fs] : fs_by_prefix_) {
+    if (prefix.size() > best_len && LocationUtil::PathHasPrefix(canonical, prefix)) {
+      best = &fs;
+      best_len = prefix.size();
+    }
+  }
+  return *best;
+}
+
+Status ArrowFileSystemFileIO::SetStorageCredentials(
+    const std::vector<
+        std::pair<std::string, std::unordered_map<std::string, std::string>>>&
+        properties_by_prefix) {
+#if ICEBERG_S3_ENABLED
+  std::vector<std::pair<std::string, std::shared_ptr<::arrow::fs::FileSystem>>>
+      fs_by_prefix;
+  fs_by_prefix.reserve(properties_by_prefix.size());
+  for (const auto& [prefix, properties] : properties_by_prefix) {
+    ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
+    fs_by_prefix.emplace_back(prefix, std::move(fs));
+  }
+  fs_by_prefix_ = std::move(fs_by_prefix);
+  return {};
+#else
+  (void)properties_by_prefix;
+  return NotSupported("S3 storage credentials require Arrow S3 support");
+#endif
+}
+
+Result<std::string> ArrowFileSystemFileIO::ResolvePath(
+    const std::shared_ptr<::arrow::fs::FileSystem>& fs,
+    const std::string& file_location) {
   const auto pos = file_location.find("://");
   if (pos == std::string::npos) {
     return file_location;
   }
 
-  auto path = arrow_fs_->PathFromUri(file_location);
+  auto path = fs->PathFromUri(file_location);
   if (path.ok()) {
     return std::move(path).ValueOrDie();
   }
@@ -502,14 +545,14 @@ Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenArrowInputStream(
   ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
 
   if (auto arrow_io = std::dynamic_pointer_cast<ArrowFileSystemFileIO>(io)) {
-    ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path));
+    const auto& fs = arrow_io->FileSystemForPath(path);
+    ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(fs, path));
     ::arrow::fs::FileInfo file_info(resolved_path, ::arrow::fs::FileType::File);
     if (length.has_value()) {
       ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(*length));
       file_info.set_size(size);
     }
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input,
-                                   arrow_io->arrow_fs_->OpenInputFile(file_info));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input, fs->OpenInputFile(file_info));
     return input;
   }
 
@@ -533,16 +576,15 @@ Result<std::shared_ptr<::arrow::io::OutputStream>> OpenArrowOutputStream(
   ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
 
   if (auto arrow_io = std::dynamic_pointer_cast<ArrowFileSystemFileIO>(io)) {
-    ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path));
+    const auto& fs = arrow_io->FileSystemForPath(path);
+    ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(fs, path));
     if (!overwrite) {
-      ICEBERG_ARROW_ASSIGN_OR_RETURN(auto info,
-                                     arrow_io->arrow_fs_->GetFileInfo(resolved_path));
+      ICEBERG_ARROW_ASSIGN_OR_RETURN(auto info, fs->GetFileInfo(resolved_path));
       if (info.type() != ::arrow::fs::FileType::NotFound) {
         return AlreadyExists("File already exists: {}", path);
       }
     }
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto output,
-                                   arrow_io->arrow_fs_->OpenOutputStream(resolved_path));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto output, fs->OpenOutputStream(resolved_path));
     return output;
   }
 
@@ -558,42 +600,60 @@ Result<std::shared_ptr<::arrow::io::OutputStream>> OpenArrowOutputStream(
 
 Result<std::unique_ptr<InputFile>> ArrowFileSystemFileIO::NewInputFile(
     std::string file_location) {
-  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
-  return std::make_unique<ArrowInputFile>(arrow_fs_, std::move(file_location),
-                                          std::move(path), std::nullopt);
+  const auto& fs = FileSystemForPath(file_location);
+  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
+  return std::make_unique<ArrowInputFile>(fs, std::move(file_location), std::move(path),
+                                          std::nullopt);
 }
 
 Result<std::unique_ptr<InputFile>> ArrowFileSystemFileIO::NewInputFile(
     std::string file_location, size_t length) {
   ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(length));
-  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
-  return std::make_unique<ArrowInputFile>(arrow_fs_, std::move(file_location),
-                                          std::move(path), size);
+  const auto& fs = FileSystemForPath(file_location);
+  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
+  return std::make_unique<ArrowInputFile>(fs, std::move(file_location), std::move(path),
+                                          size);
 }
 
 Result<std::unique_ptr<OutputFile>> ArrowFileSystemFileIO::NewOutputFile(
     std::string file_location) {
-  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
-  return std::make_unique<ArrowOutputFile>(arrow_fs_, std::move(file_location),
-                                           std::move(path));
+  const auto& fs = FileSystemForPath(file_location);
+  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
+  return std::make_unique<ArrowOutputFile>(fs, std::move(file_location), std::move(path));
 }
 
 /// \brief Delete a file at the given location.
 Status ArrowFileSystemFileIO::DeleteFile(const std::string& file_location) {
-  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
-  ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFile(path));
+  const auto& fs = FileSystemForPath(file_location);
+  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
+  ICEBERG_ARROW_RETURN_NOT_OK(fs->DeleteFile(path));
   return {};
 }
 
 Status ArrowFileSystemFileIO::DeleteFiles(
     const std::vector<std::string>& file_locations) {
-  std::vector<std::string> paths;
-  paths.reserve(file_locations.size());
+  if (fs_by_prefix_.empty()) {
+    // No per-prefix routing: one batched delete on the default file system.
+    std::vector<std::string> paths;
+    paths.reserve(file_locations.size());
+    for (const auto& file_location : file_locations) {
+      ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(arrow_fs_, file_location));
+      paths.push_back(std::move(path));
+    }
+    ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFiles(paths));
+    return {};
+  }
+
+  // Paths may route to different file systems: group by fs, then batch per fs.
+  std::unordered_map<::arrow::fs::FileSystem*, std::vector<std::string>> paths_by_fs;
   for (const auto& file_location : file_locations) {
-    ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
-    paths.push_back(std::move(path));
+    const auto& fs = FileSystemForPath(file_location);
+    ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
+    paths_by_fs[fs.get()].push_back(std::move(path));
+  }
+  for (auto& [fs, paths] : paths_by_fs) {
+    ICEBERG_ARROW_RETURN_NOT_OK(fs->DeleteFiles(paths));
   }
-  ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFiles(paths));
   return {};
 }
 
diff --git a/src/iceberg/arrow/arrow_io_internal.h b/src/iceberg/arrow/arrow_io_internal.h
index a6b85b6c9..0e3607cb2 100644
--- a/src/iceberg/arrow/arrow_io_internal.h
+++ b/src/iceberg/arrow/arrow_io_internal.h
@@ -23,6 +23,9 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <string_view>
+#include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include <arrow/filesystem/type_fwd.h>
@@ -52,7 +55,8 @@ OpenArrowOutputStream(const std::shared_ptr<FileIO>& io, const std::string& path
                       bool overwrite = true);
 
 /// \brief A concrete implementation of FileIO for Arrow file system.
-class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
+class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO,
+                                                    public SupportsStorageCredentials {
  public:
   explicit ArrowFileSystemFileIO(std::shared_ptr<::arrow::fs::FileSystem> arrow_fs)
       : arrow_fs_(std::move(arrow_fs)) {}
@@ -81,6 +85,12 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
   /// \brief Delete files at the given locations.
   Status DeleteFiles(const std::vector<std::string>& file_locations) override;
 
+  /// \brief Build one S3 file system per credential prefix for per-path routing.
+  Status SetStorageCredentials(
+      const std::vector<
+          std::pair<std::string, std::unordered_map<std::string, std::string>>>&
+          properties_by_prefix) override;
+
   /// \brief Get the Arrow file system.
   const std::shared_ptr<::arrow::fs::FileSystem>& fs() const { return arrow_fs_; }
 
@@ -92,10 +102,18 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
   friend Result<std::shared_ptr<::arrow::io::OutputStream>> OpenArrowOutputStream(
       const std::shared_ptr<FileIO>& io, const std::string& path, bool overwrite);
 
-  /// \brief Resolve a file location to a filesystem path.
-  Result<std::string> ResolvePath(const std::string& file_location);
+  /// \brief Pick the file system for `location` (longest matching prefix, else
+  /// the default).
+  const std::shared_ptr<::arrow::fs::FileSystem>& FileSystemForPath(
+      std::string_view location) const;
+
+  /// \brief Resolve a file location to a filesystem path using `fs`.
+  Result<std::string> ResolvePath(const std::shared_ptr<::arrow::fs::FileSystem>& fs,
+                                  const std::string& file_location);
 
   std::shared_ptr<::arrow::fs::FileSystem> arrow_fs_;
+  std::vector<std::pair<std::string, std::shared_ptr<::arrow::fs::FileSystem>>>
+      fs_by_prefix_;
 };
 
 }  // namespace iceberg::arrow
diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc
index cffd95840..f65f023ca 100644
--- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc
+++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc
@@ -21,6 +21,7 @@
 #include <optional>
 #include <string>
 #include <string_view>
+#include <utility>
 
 #include <arrow/filesystem/filesystem.h>
 #if ICEBERG_S3_ENABLED
@@ -30,6 +31,7 @@
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_io_util.h"
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow/s3/arrow_s3_internal.h"
 #include "iceberg/arrow/s3/s3_properties.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/string_util.h"
@@ -74,6 +76,23 @@ Status EnsureS3Initialized() {
   return {};
 }
 
+// Splits any URI scheme off `endpoint` into `options.scheme`, returning the bare
+// host[:port] that Arrow's `endpoint_override` expects.
+std::string SplitEndpointScheme(std::string_view endpoint,
+                                ::arrow::fs::S3Options& options) {
+  if (const auto pos = endpoint.find("://"); pos != std::string_view::npos) {
+    options.scheme = std::string(endpoint.substr(0, pos));
+    endpoint = endpoint.substr(pos + 3);
+  }
+  return std::string(endpoint);
+}
+
+#endif
+
+}  // namespace
+
+#if ICEBERG_S3_ENABLED
+
 /// \brief Configure S3Options from a properties map.
 ///
 /// \param properties The configuration properties map.
@@ -100,26 +119,26 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
   }
 
   // Configure region
-  if (const auto* region = FindProperty(properties, S3Properties::kRegion);
-      region != nullptr) {
+  // Prefer the standard `client.region`; fall back to legacy `s3.region`.
+  const auto* region = FindProperty(properties, S3Properties::kClientRegion);
+  if (region == nullptr) {
+    region = FindProperty(properties, S3Properties::kRegion);
+  }
+  if (region != nullptr) {
     options.region = *region;
   }
 
-  // Configure endpoint (for MinIO, LocalStack, etc.)
+  // Configure endpoint (for MinIO, LocalStack, OSS, etc.) from `s3.endpoint` or
+  // the AWS standard env vars.
   if (const auto* endpoint = FindProperty(properties, S3Properties::kEndpoint);
       endpoint != nullptr) {
-    options.endpoint_override = *endpoint;
-  } else {
-    // Fall back to AWS standard environment variables for endpoint override
-    const char* s3_endpoint_env = std::getenv("AWS_ENDPOINT_URL_S3");
-    if (s3_endpoint_env != nullptr) {
-      options.endpoint_override = s3_endpoint_env;
-    } else {
-      const char* endpoint_env = std::getenv("AWS_ENDPOINT_URL");
-      if (endpoint_env != nullptr) {
-        options.endpoint_override = endpoint_env;
-      }
-    }
+    options.endpoint_override = SplitEndpointScheme(*endpoint, options);
+  } else if (const char* s3_endpoint_env = std::getenv("AWS_ENDPOINT_URL_S3");
+             s3_endpoint_env != nullptr) {
+    options.endpoint_override = SplitEndpointScheme(s3_endpoint_env, options);
+  } else if (const char* endpoint_env = std::getenv("AWS_ENDPOINT_URL");
+             endpoint_env != nullptr) {
+    options.endpoint_override = SplitEndpointScheme(endpoint_env, options);
   }
 
   ICEBERG_ASSIGN_OR_RAISE(const auto path_style_access,
@@ -128,11 +147,11 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
     options.force_virtual_addressing = !*path_style_access;
   }
 
-  // Configure SSL
+  // Explicit `s3.ssl.enabled` overrides any endpoint-derived scheme.
   ICEBERG_ASSIGN_OR_RAISE(const auto ssl_enabled,
                           ParseOptionalBool(properties, S3Properties::kSslEnabled));
-  if (ssl_enabled.has_value() && !*ssl_enabled) {
-    options.scheme = "http";
+  if (ssl_enabled.has_value()) {
+    options.scheme = *ssl_enabled ? "https" : "http";
   }
 
   // Configure timeouts
@@ -154,17 +173,21 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
 }
 #endif
 
-}  // namespace
-
-Result<std::unique_ptr<FileIO>> MakeS3FileIO(
-    const std::unordered_map<std::string, std::string>& properties) {
 #if ICEBERG_S3_ENABLED
+Result<std::shared_ptr<::arrow::fs::FileSystem>> BuildArrowS3FileSystem(
+    const std::unordered_map<std::string, std::string>& properties) {
   ICEBERG_RETURN_UNEXPECTED(EnsureS3Initialized());
-
-  // Configure S3 options from properties (uses default credentials if empty)
   ICEBERG_ASSIGN_OR_RAISE(auto options, ConfigureS3Options(properties));
   ICEBERG_ARROW_ASSIGN_OR_RETURN(auto fs, ::arrow::fs::S3FileSystem::Make(options));
+  return std::shared_ptr<::arrow::fs::FileSystem>(std::move(fs));
+}
+#endif
 
+Result<std::unique_ptr<FileIO>> MakeS3FileIO(
+    const std::unordered_map<std::string, std::string>& properties) {
+#if ICEBERG_S3_ENABLED
+  // Uses default credentials if properties are empty.
+  ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
   return std::make_unique<ArrowFileSystemFileIO>(std::move(fs));
 #else
   return NotSupported("Arrow S3 support is not enabled");
diff --git a/src/iceberg/arrow/s3/arrow_s3_internal.h b/src/iceberg/arrow/s3/arrow_s3_internal.h
new file mode 100644
index 000000000..347c879e7
--- /dev/null
+++ b/src/iceberg/arrow/s3/arrow_s3_internal.h
@@ -0,0 +1,50 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/iceberg_bundle_export.h"
+#include "iceberg/result.h"
+
+#if ICEBERG_S3_ENABLED
+#  include <arrow/filesystem/s3fs.h>
+#endif
+
+namespace iceberg::arrow {
+
+#if ICEBERG_S3_ENABLED
+/// \brief Build Arrow ``S3Options`` from an Iceberg properties map.
+///
+/// Production code should use MakeS3FileIO(); this is exposed so the
+/// property-to-option mapping (region resolution, endpoint scheme handling,
+/// addressing style) can be unit tested without a live S3 endpoint.
+ICEBERG_BUNDLE_EXPORT Result<::arrow::fs::S3Options> ConfigureS3Options(
+    const std::unordered_map<std::string, std::string>& properties);
+
+/// \brief Build an Arrow S3 file system from a properties map (initializes S3 if
+/// needed). Exposed so the credential-aware FileIO can build one fs per prefix.
+ICEBERG_BUNDLE_EXPORT Result<std::shared_ptr<::arrow::fs::FileSystem>>
+BuildArrowS3FileSystem(const std::unordered_map<std::string, std::string>& properties);
+#endif
+
+}  // namespace iceberg::arrow
diff --git a/src/iceberg/arrow/s3/s3_properties.h b/src/iceberg/arrow/s3/s3_properties.h
index 53657743d..61248948b 100644
--- a/src/iceberg/arrow/s3/s3_properties.h
+++ b/src/iceberg/arrow/s3/s3_properties.h
@@ -37,8 +37,10 @@ struct S3Properties {
   static constexpr std::string_view kSecretAccessKey = "s3.secret-access-key";
   /// AWS session token (for temporary credentials)
   static constexpr std::string_view kSessionToken = "s3.session-token";
-  /// AWS region
+  /// AWS region (legacy, non-standard key kept for compatibility)
   static constexpr std::string_view kRegion = "s3.region";
+  /// AWS region, standard Iceberg client property (preferred over kRegion).
+  static constexpr std::string_view kClientRegion = "client.region";
   /// Custom endpoint override (for MinIO, LocalStack, etc.)
   static constexpr std::string_view kEndpoint = "s3.endpoint";
   /// Whether to use path-style access (needed for MinIO)
diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc
index ac80e9f08..30ab6de03 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -71,6 +71,8 @@ constexpr std::string_view kSource = "source";
 constexpr std::string_view kDestination = "destination";
 constexpr std::string_view kMetadata = "metadata";
 constexpr std::string_view kConfig = "config";
+constexpr std::string_view kStorageCredentials = "storage-credentials";
+constexpr std::string_view kPrefix = "prefix";
 constexpr std::string_view kIdentifiers = "identifiers";
 constexpr std::string_view kOverrides = "overrides";
 constexpr std::string_view kDefaults = "defaults";
@@ -695,6 +697,17 @@ nlohmann::json ToJson(const LoadTableResult& result) {
   SetOptionalStringField(json, kMetadataLocation, result.metadata_location);
   json[kMetadata] = ToJson(*result.metadata);
   SetContainerField(json, kConfig, result.config);
+  if (!result.storage_credentials.empty()) {
+    nlohmann::json creds = nlohmann::json::array();
+    for (const auto& cred : result.storage_credentials) {
+      nlohmann::json entry;
+      entry[kPrefix] = cred.prefix;
+      // config is required, so always write it (matches Java).
+      entry[kConfig] = cred.config;
+      creds.push_back(std::move(entry));
+    }
+    json[kStorageCredentials] = std::move(creds);
+  }
   return json;
 }
 
@@ -707,6 +720,27 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
   ICEBERG_ASSIGN_OR_RAISE(result.config,
                           GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
+  if (auto it = json.find(kStorageCredentials); it != json.end() && !it->is_null()) {
+    if (!it->is_array()) {
+      // Don't echo the value — it may carry credential material.
+      return JsonParseError("Cannot parse storage credentials from non-array");
+    }
+    for (const auto& entry : *it) {
+      StorageCredential cred;
+      ICEBERG_ASSIGN_OR_RAISE(cred.prefix, GetJsonValue<std::string>(entry, kPrefix));
+      ICEBERG_ASSIGN_OR_RAISE(cred.config,
+                              GetJsonValue<decltype(cred.config)>(entry, kConfig));
+      // prefix and config are required by the REST spec; non-empty matches the
+      // Java reference implementation (Credential.validate()).
+      if (cred.prefix.empty()) {
+        return JsonParseError("Invalid storage credential: prefix must be non-empty");
+      }
+      if (cred.config.empty()) {
+        return JsonParseError("Invalid storage credential: config must be non-empty");
+      }
+      result.storage_credentials.push_back(std::move(cred));
+    }
+  }
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
   return result;
 }
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
index 455c4d744..8f786995f 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -20,6 +20,7 @@
 #include "iceberg/catalog/rest/rest_catalog.h"
 
 #include <memory>
+#include <ranges>
 #include <string_view>
 #include <tuple>
 #include <unordered_map>
@@ -50,6 +51,7 @@
 #include "iceberg/table_requirement.h"
 #include "iceberg/table_update.h"
 #include "iceberg/transaction.h"
+#include "iceberg/util/formatter_internal.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -284,12 +286,14 @@ class RestCatalog::TableScopedCatalog final
   TableScopedCatalog(std::shared_ptr<RestCatalog> root, SessionContext context,
                      TableIdentifier identifier,
                      std::unordered_map<std::string, std::string> table_config,
-                     std::shared_ptr<auth::AuthSession> table_session)
+                     std::shared_ptr<auth::AuthSession> table_session,
+                     std::shared_ptr<FileIO> table_io)
       : root_(std::move(root)),
         context_(std::move(context)),
         identifier_(std::move(identifier)),
         table_config_(std::move(table_config)),
-        table_session_(std::move(table_session)) {}
+        table_session_(std::move(table_session)),
+        table_io_(std::move(table_io)) {}
 
   std::string_view name() const override { return root_->name(); }
 
@@ -347,7 +351,7 @@ class RestCatalog::TableScopedCatalog final
         auto response,
         root_->UpdateTableInternal(identifier, requirements, updates, *table_session_));
     return root_->MakeTableFromCommitResponse(identifier, std::move(response), context_,
-                                              table_config_, table_session_);
+                                              table_config_, table_session_, table_io_);
   }
 
   Result<std::shared_ptr<Transaction>> StageCreateTable(
@@ -394,6 +398,7 @@ class RestCatalog::TableScopedCatalog final
   TableIdentifier identifier_;
   std::unordered_map<std::string, std::string> table_config_;
   std::shared_ptr<auth::AuthSession> table_session_;
+  std::shared_ptr<FileIO> table_io_;
 };
 
 RestCatalog::~RestCatalog() {
@@ -516,7 +521,26 @@ 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) const {
+    const std::unordered_map<std::string, std::string>& table_config,
+    const std::vector<StorageCredential>& storage_credentials) const {
+  if (!storage_credentials.empty()) {
+    // Only non-S3 (GCS/ADLS) credentials vended, which we can't honor; fail fast.
+    if (HasOnlyNonS3StorageCredentials(storage_credentials)) {
+      auto prefixes =
+          storage_credentials | std::views::transform(&StorageCredential::prefix);
+      return NotSupported(
+          "Vended storage credentials {} are unsupported (only S3-family "
+          "credentials are supported)",
+          FormatRange(prefixes, ", ", "[", "]"));
+    }
+    // Build a FileIO that routes each path to its vended credential. TODO: STS
+    // refresh via credentials.uri is not yet supported.
+    ICEBERG_ASSIGN_OR_RAISE(auto io,
+                            MakeS3FileIOFromCredentials(config_.configs(), table_config,
+                                                        storage_credentials));
+    return std::shared_ptr<FileIO>(std::move(io));
+  }
+
   ICEBERG_RETURN_UNEXPECTED(ValidateNoFileIOConfig(table_config));
   return file_io_;
 }
@@ -730,8 +754,10 @@ Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
+  // No table was loaded here, so there is no per-table FileIO to reuse; use the
+  // catalog FileIO (table_config carries no credentials).
   return MakeTableFromCommitResponse(identifier, std::move(response), context,
-                                     table_config, std::move(table_session));
+                                     table_config, std::move(table_session), file_io_);
 }
 
 Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
@@ -746,12 +772,15 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
       CreateTableInternal(identifier, schema, spec, order, location, properties,
                           /*stage_create=*/true, *contextual_session));
   auto table_config = std::move(result.config);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, table_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_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
-      shared_from_this(), context, identifier, table_config, std::move(table_session));
+      shared_from_this(), context, identifier, table_config, std::move(table_session),
+      table_io);
   ICEBERG_ASSIGN_OR_RAISE(
       auto staged_table,
       StagedTable::Make(identifier, std::move(result.metadata),
@@ -859,12 +888,14 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
     const SessionContext& context,
     std::shared_ptr<auth::AuthSession> contextual_session) {
   auto table_config = std::move(result.config);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, table_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_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
-      shared_from_this(), context, identifier, table_config, table_session);
+      shared_from_this(), context, identifier, table_config, table_session, table_io);
   return Table::Make(identifier, std::move(result.metadata),
                      std::move(result.metadata_location), std::move(table_io),
                      std::move(table_catalog));
@@ -874,13 +905,14 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromCommitResponse(
     const TableIdentifier& identifier, CommitTableResponse response,
     const SessionContext& context,
     const std::unordered_map<std::string, std::string>& table_config,
-    std::shared_ptr<auth::AuthSession> table_session) {
-  // TODO(gangwu): If the REST commit response grows table config or
-  // storage credentials, derive a replacement table session/FileIO from that
-  // response. The current table commit response does not define config.
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, table_config));
+    std::shared_ptr<auth::AuthSession> table_session, std::shared_ptr<FileIO> table_io) {
+  // Reuse the FileIO bound at load: CommitTableResponse carries no config or
+  // storage credentials, so rebuilding it would drop the vended credentials
+  // (mirrors Java RESTSessionCatalog#tableFileIO).
+  // TODO(gangwu): rebuild the FileIO if the commit response ever grows config
+  // or storage credentials.
   auto table_catalog = std::make_shared<TableScopedCatalog>(
-      shared_from_this(), context, identifier, table_config, table_session);
+      shared_from_this(), context, identifier, table_config, table_session, table_io);
   return Table::Make(identifier, std::move(response.metadata),
                      std::move(response.metadata_location), std::move(table_io),
                      std::move(table_catalog));
diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h
index e693ceba3..ed1129caa 100644
--- a/src/iceberg/catalog/rest/rest_catalog.h
+++ b/src/iceberg/catalog/rest/rest_catalog.h
@@ -79,7 +79,8 @@ 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) const;
+      const std::unordered_map<std::string, std::string>& table_config,
+      const std::vector<StorageCredential>& storage_credentials) const;
 
   Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns,
                                                 auth::AuthSession& session) const;
@@ -169,7 +170,7 @@ class ICEBERG_REST_EXPORT RestCatalog final
       const TableIdentifier& identifier, CommitTableResponse response,
       const SessionContext& context,
       const std::unordered_map<std::string, std::string>& table_config,
-      std::shared_ptr<auth::AuthSession> table_session);
+      std::shared_ptr<auth::AuthSession> table_session, std::shared_ptr<FileIO> table_io);
 
   RestCatalogProperties config_;
   std::shared_ptr<FileIO> file_io_;
diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc
index 5fadca1ac..e0f47613a 100644
--- a/src/iceberg/catalog/rest/rest_file_io.cc
+++ b/src/iceberg/catalog/rest/rest_file_io.cc
@@ -19,9 +19,16 @@
 
 #include "iceberg/catalog/rest/rest_file_io.h"
 
+#include <algorithm>
 #include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
 
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
+#include "iceberg/util/location_util.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -92,4 +99,49 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
   return FileIORegistry::Load(io_impl, config.configs());
 }
 
+bool HasOnlyNonS3StorageCredentials(const std::vector<StorageCredential>& credentials) {
+  return !credentials.empty() &&
+         std::ranges::none_of(credentials, [](const StorageCredential& cred) {
+           return cred.prefix.starts_with("s3");
+         });
+}
+
+Result<std::unique_ptr<FileIO>> MakeS3FileIOFromCredentials(
+    const std::unordered_map<std::string, std::string>& catalog_config,
+    const std::unordered_map<std::string, std::string>& table_config,
+    const std::vector<StorageCredential>& storage_credentials) {
+  auto default_properties = catalog_config;
+  for (const auto& [key, value] : table_config) {
+    default_properties[key] = value;
+  }
+
+  // Default S3 FileIO (for paths matching no credential prefix), built via the
+  // registry to keep this layer decoupled from the Arrow/S3 implementation.
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto io, FileIORegistry::Load(std::string(FileIORegistry::kArrowS3FileIO),
+                                    default_properties));
+
+  // One property set per S3-family credential, keyed by canonicalized prefix;
+  // the credential's config overrides the default properties.
+  std::vector<std::pair<std::string, std::unordered_map<std::string, std::string>>>
+      properties_by_prefix;
+  for (const auto& cred : storage_credentials) {
+    if (!cred.prefix.starts_with("s3")) {
+      continue;
+    }
+    auto properties = default_properties;
+    for (const auto& [key, value] : cred.config) {
+      properties[key] = value;
+    }
+    properties_by_prefix.emplace_back(LocationUtil::CanonicalizeS3Scheme(cred.prefix),
+                                      std::move(properties));
+  }
+
+  // Hand the per-prefix properties to the FileIO for per-path credential routing.
+  if (auto* credentialed = dynamic_cast<SupportsStorageCredentials*>(io.get())) {
+    ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(properties_by_prefix));
+  }
+  return io;
+}
+
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/rest_file_io.h b/src/iceberg/catalog/rest/rest_file_io.h
index 68482521a..d5052143f 100644
--- a/src/iceberg/catalog/rest/rest_file_io.h
+++ b/src/iceberg/catalog/rest/rest_file_io.h
@@ -22,9 +22,12 @@
 #include <cstdint>
 #include <memory>
 #include <string_view>
+#include <unordered_map>
+#include <vector>
 
 #include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/result.h"
@@ -44,4 +47,17 @@ ICEBERG_REST_EXPORT std::string_view BuiltinFileIOName(BuiltinFileIOKind kind);
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
     const RestCatalogProperties& config);
 
+/// \brief True if `credentials` is non-empty but has no S3-family credential
+/// (prefix starting with "s3") — only unsupported schemes (GCS/ADLS) were vended.
+ICEBERG_REST_EXPORT bool HasOnlyNonS3StorageCredentials(
+    const std::vector<StorageCredential>& credentials);
+
+/// \brief Build an S3 FileIO that routes each object path to a per-prefix file
+/// system, one per S3-family vended credential (config merged catalog < table <
+/// credential). Non-S3 credentials are ignored.
+ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeS3FileIOFromCredentials(
+    const std::unordered_map<std::string, std::string>& catalog_config,
+    const std::unordered_map<std::string, std::string>& table_config,
+    const std::vector<StorageCredential>& storage_credentials);
+
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/type_fwd.h b/src/iceberg/catalog/rest/type_fwd.h
index ee684b245..783f22750 100644
--- a/src/iceberg/catalog/rest/type_fwd.h
+++ b/src/iceberg/catalog/rest/type_fwd.h
@@ -28,6 +28,7 @@ struct ErrorResponse;
 struct CommitTableResponse;
 struct LoadTableResult;
 struct OAuthTokenResponse;
+struct StorageCredential;
 
 class Endpoint;
 class ErrorHandler;
diff --git a/src/iceberg/catalog/rest/types.cc b/src/iceberg/catalog/rest/types.cc
index 8d96bccb2..84fba9a7c 100644
--- a/src/iceberg/catalog/rest/types.cc
+++ b/src/iceberg/catalog/rest/types.cc
@@ -86,7 +86,8 @@ bool CreateTableRequest::operator==(const CreateTableRequest& other) const {
 }
 
 bool LoadTableResult::operator==(const LoadTableResult& other) const {
-  if (metadata_location != other.metadata_location || config != other.config) {
+  if (metadata_location != other.metadata_location || config != other.config ||
+      storage_credentials != other.storage_credentials) {
     return false;
   }
 
diff --git a/src/iceberg/catalog/rest/types.h b/src/iceberg/catalog/rest/types.h
index 7849b366b..3bf218a6f 100644
--- a/src/iceberg/catalog/rest/types.h
+++ b/src/iceberg/catalog/rest/types.h
@@ -180,18 +180,44 @@ struct ICEBERG_REST_EXPORT CreateTableRequest {
 /// \brief An opaque token that allows clients to make use of pagination for list APIs.
 using PageToken = std::string;
 
+/// \brief A short-lived credential vended by a REST catalog for a storage
+/// location ``prefix`` (clients pick the longest matching prefix); ``config``
+/// holds backend properties such as ``"s3.access-key-id"`` (Iceberg REST spec).
+struct ICEBERG_REST_EXPORT StorageCredential {
+  std::string prefix;
+  std::unordered_map<std::string, std::string> config;
+
+  /// \brief Validates the StorageCredential. The REST spec requires both a
+  /// prefix and config; non-empty matches Java `Credential.validate()`.
+  Status Validate() const {
+    if (prefix.empty()) {
+      return ValidationFailed("Invalid storage credential: prefix must be non-empty");
+    }
+    if (config.empty()) {
+      return ValidationFailed("Invalid storage credential: config must be non-empty");
+    }
+    return {};
+  }
+
+  bool operator==(const StorageCredential& other) const = default;
+};
+
 /// \brief Result body for table create/load/register APIs.
 struct ICEBERG_REST_EXPORT LoadTableResult {
   std::string metadata_location;
   std::shared_ptr<TableMetadata> metadata;  // required
   std::unordered_map<std::string, std::string> config;
-  // TODO(Li Feiyang): Add std::shared_ptr<StorageCredential> storage_credential;
+  /// \brief Vended storage credentials, one per URI prefix; empty if none.
+  std::vector<StorageCredential> storage_credentials;
 
   /// \brief Validates the LoadTableResult.
   Status Validate() const {
     if (!metadata) {
       return ValidationFailed("Invalid metadata: null");
     }
+    for (const auto& credential : storage_credentials) {
+      ICEBERG_RETURN_UNEXPECTED(credential.Validate());
+    }
     return {};
   }
 
diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h
index 1f91fb0c1..a760b94d4 100644
--- a/src/iceberg/file_io.h
+++ b/src/iceberg/file_io.h
@@ -26,6 +26,8 @@
 #include <span>
 #include <string>
 #include <string_view>
+#include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
@@ -167,4 +169,19 @@ class ICEBERG_EXPORT FileIO {
   virtual Status DeleteFiles(const std::vector<std::string>& file_locations);
 };
 
+/// \brief Mix-in for FileIO implementations that route object paths to
+/// per-prefix file systems built from vended storage credentials, letting the
+/// catalog stay decoupled from the concrete (Arrow/S3) implementation.
+class ICEBERG_EXPORT SupportsStorageCredentials {
+ public:
+  virtual ~SupportsStorageCredentials() = default;
+
+  /// \brief Install per-prefix file systems. `properties_by_prefix` maps a
+  /// canonicalized prefix to its fully merged properties.
+  virtual Status SetStorageCredentials(
+      const std::vector<
+          std::pair<std::string, std::unordered_map<std::string, std::string>>>&
+          properties_by_prefix) = 0;
+};
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc
index b1caff1e8..ef8291050 100644
--- a/src/iceberg/test/arrow_s3_file_io_test.cc
+++ b/src/iceberg/test/arrow_s3_file_io_test.cc
@@ -27,6 +27,7 @@
 #include <gtest/gtest.h>
 
 #include "iceberg/arrow/arrow_io_util.h"
+#include "iceberg/arrow/s3/arrow_s3_internal.h"
 #include "iceberg/arrow/s3/s3_properties.h"
 #include "iceberg/test/matchers.h"
 
@@ -76,6 +77,13 @@ namespace {
 
 class ArrowS3FileIOTest : public ::testing::Test {
  protected:
+#if ICEBERG_S3_ENABLED
+  static void SetUpTestSuite() {
+    auto io = MakeS3FileIO({});
+    ASSERT_THAT(io, IsOk());
+  }
+#endif
+
   static void TearDownTestSuite() {
     auto status = FinalizeS3();
     if (!status.has_value()) {
@@ -181,4 +189,73 @@ TEST_F(ArrowS3FileIOTest, MakeS3FileIOWithTimeouts) {
   ASSERT_THAT(io_res, IsOk());
 }
 
+#if ICEBERG_S3_ENABLED
+TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsPrefersClientRegionOverS3Region) {
+  auto result =
+      ConfigureS3Options({{std::string(S3Properties::kClientRegion), "cn-hangzhou"},
+                          {std::string(S3Properties::kRegion), "us-east-1"}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_EQ(result->region, "cn-hangzhou");
+}
+
+TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsFallsBackToS3Region) {
+  auto result = ConfigureS3Options({{std::string(S3Properties::kRegion), "us-east-1"}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_EQ(result->region, "us-east-1");
+}
+
+TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsStripsHttpsEndpointScheme) {
+  auto result = ConfigureS3Options({{std::string(S3Properties::kEndpoint),
+                                     "https://oss-cn-hangzhou.aliyuncs.com:443"}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_EQ(result->endpoint_override, "oss-cn-hangzhou.aliyuncs.com:443");
+  EXPECT_EQ(result->scheme, "https");
+}
+
+TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsStripsHttpEndpointScheme) {
+  auto result = ConfigureS3Options(
+      {{std::string(S3Properties::kEndpoint), "http://localhost:9000"}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_EQ(result->endpoint_override, "localhost:9000");
+  EXPECT_EQ(result->scheme, "http");
+}
+
+TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsKeepsSchemelessEndpoint) {
+  auto result =
+      ConfigureS3Options({{std::string(S3Properties::kEndpoint), "localhost:9000"}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_EQ(result->endpoint_override, "localhost:9000");
+}
+
+TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsSslEnabledOverridesEndpointScheme) {
+  auto https =
+      ConfigureS3Options({{std::string(S3Properties::kEndpoint), "http://localhost:9000"},
+                          {std::string(S3Properties::kSslEnabled), "true"}});
+  ASSERT_THAT(https, IsOk());
+  EXPECT_EQ(https->scheme, "https");
+
+  auto http = ConfigureS3Options(
+      {{std::string(S3Properties::kEndpoint), "https://localhost:9000"},
+       {std::string(S3Properties::kSslEnabled), "false"}});
+  ASSERT_THAT(http, IsOk());
+  EXPECT_EQ(http->scheme, "http");
+}
+
+TEST_F(ArrowS3FileIOTest,
+       ConfigureS3OptionsPathStyleAccessFalseEnablesVirtualAddressing) {
+  auto result =
+      ConfigureS3Options({{std::string(S3Properties::kPathStyleAccess), "false"}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_TRUE(result->force_virtual_addressing);
+}
+
+TEST_F(ArrowS3FileIOTest,
+       ConfigureS3OptionsPathStyleAccessTrueDisablesVirtualAddressing) {
+  auto result =
+      ConfigureS3Options({{std::string(S3Properties::kPathStyleAccess), "true"}});
+  ASSERT_THAT(result, IsOk());
+  EXPECT_FALSE(result->force_virtual_addressing);
+}
+#endif
+
 }  // namespace iceberg::arrow
diff --git a/src/iceberg/test/rest_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc
index b1193d9f8..d54943fe4 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -19,11 +19,17 @@
 
 #include "iceberg/catalog/rest/rest_file_io.h"
 
+#include <memory>
+#include <unordered_map>
+#include <vector>
+
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io_registry.h"
 #include "iceberg/test/matchers.h"
+#include "iceberg/util/location_util.h"
 
 namespace iceberg::rest {
 
@@ -147,4 +153,46 @@ TEST(RestFileIOTest, MakeCatalogFileIOSkipsCheckWhenWarehouseAbsent) {
   ASSERT_THAT(result, IsOk());
 }
 
+TEST(RestFileIOTest, CanonicalizeS3SchemeTreatsS3CompatibleSchemesAsS3) {
+  // s3a/s3n/oss canonicalize to s3:// so a vended `s3` credential prefix-matches
+  // them uniformly (DLF vends `s3` for oss:// locations).
+  EXPECT_EQ(LocationUtil::CanonicalizeS3Scheme("oss://bucket/db/t"), "s3://bucket/db/t");
+  EXPECT_EQ(LocationUtil::CanonicalizeS3Scheme("s3a://bucket/x"), "s3://bucket/x");
+  EXPECT_EQ(LocationUtil::CanonicalizeS3Scheme("s3n://bucket/x"), "s3://bucket/x");
+  // Already-canonical and non-S3 / scheme-less locations are unchanged.
+  EXPECT_EQ(LocationUtil::CanonicalizeS3Scheme("s3://bucket/x"), "s3://bucket/x");
+  EXPECT_EQ(LocationUtil::CanonicalizeS3Scheme("gs://bucket"), "gs://bucket");
+  EXPECT_EQ(LocationUtil::CanonicalizeS3Scheme("/local/path"), "/local/path");
+}
+
+TEST(RestFileIOTest, PathHasPrefixMatchesAtPathBoundary) {
+  // Must match only at a path boundary, so a `s3://bucket/db/t1` credential does
+  // not capture a sibling table under `s3://bucket/db/t1x/...`.
+  EXPECT_TRUE(LocationUtil::PathHasPrefix("s3://bucket/db/t1/data/f.parquet",
+                                          "s3://bucket/db/t1"));
+  EXPECT_TRUE(LocationUtil::PathHasPrefix("s3://bucket/db/t1", "s3://bucket/db/t1"));
+  EXPECT_FALSE(LocationUtil::PathHasPrefix("s3://bucket/db/t1x/data/f.parquet",
+                                           "s3://bucket/db/t1"));
+  EXPECT_FALSE(LocationUtil::PathHasPrefix("s3://bucket-other/x", "s3://bucket"));
+
+  // A bare-scheme credential (DLF vends `s3`) matches any authority/path.
+  EXPECT_TRUE(LocationUtil::PathHasPrefix("s3://bucket/db/t/f", "s3"));
+  EXPECT_TRUE(LocationUtil::PathHasPrefix(
+      LocationUtil::CanonicalizeS3Scheme("oss://bucket/db/t/f"), "s3"));
+  EXPECT_FALSE(LocationUtil::PathHasPrefix("gs://bucket/x", "s3"));
+}
+
+TEST(RestFileIOTest, HasOnlyNonS3StorageCredentials) {
+  // Only GCS/ADLS prefixes -> unsupported, fail fast.
+  EXPECT_TRUE(HasOnlyNonS3StorageCredentials(
+      {{.prefix = "gs://bucket", .config = {{"k", "v"}}},
+       {.prefix = "abfs://c@a.dfs.core.windows.net", .config = {{"k", "v"}}}}));
+  // At least one S3 credential present -> not unsupported (may fall back).
+  EXPECT_FALSE(HasOnlyNonS3StorageCredentials(
+      {{.prefix = "gs://bucket", .config = {{"k", "v"}}},
+       {.prefix = "s3://bucket", .config = {{"s3.access-key-id", "a"}}}}));
+  // No credentials at all -> not "only non-S3".
+  EXPECT_FALSE(HasOnlyNonS3StorageCredentials({}));
+}
+
 }  // namespace iceberg::rest
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
index 7304831c6..6d78a0f4a 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -82,6 +82,11 @@ static std::shared_ptr<TableMetadata> MakeSimpleTableMetadata() {
   });
 }
 
+std::string LoadTableJsonWithCredentials(std::string_view storage_credentials) {
+  return std::string(R"({"storage-credentials":)") + std::string(storage_credentials) +
+         R"(,"metadata":{"format-version":2,"table-uuid":"test","location":"s3://test","last-sequence-number":0,"last-column-id":1,"last-updated-ms":0,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0}})";
+}
+
 // Test parameter structure for roundtrip tests
 template <typename Model>
 struct JsonRoundTripParam {
@@ -1116,7 +1121,17 @@ INSTANTIATE_TEST_SUITE_P(
             .model = {.metadata_location = "s3://bucket/metadata/v1.json",
                       .metadata = MakeSimpleTableMetadata(),
                       .config = {{"warehouse", "s3://bucket/warehouse"},
-                                 {"foo", "bar"}}}}),
+                                 {"foo", "bar"}}}},
+        LoadTableResultParam{
+            .test_name = "WithStorageCredentials",
+            .expected_json_str =
+                R"({"metadata":{"current-schema-id":1,"current-snapshot-id":null,"default-sort-order-id":0,"default-spec-id":0,"format-version":2,"last-column-id":1,"last-partition-id":0,"last-sequence-number":0,"last-updated-ms":0,"location":"s3://bucket/test","metadata-log":[],"partition-specs":[{"fields":[],"spec-id":0}],"partition-statistics":[],"properties":{},"refs":{},"schemas":[{"fields":[{"id":1,"name":"id","required":true,"type":"int"}],"schema-id":1,"type":"struct"}],"snapshot-log":[],"snapshots":[],"sort-orders":[{"fields":[],"order-id":0}],"statistics":[],"table-uuid":"test-uuid-1234"},"storage-credentials":[{"config":{"s3.access-key-id":"AKIAtest","s3.region":"us-east-1","s3.secret-access-key":"secret"},"prefix":"s3"}]})",
+            .model =
+                {.metadata = MakeSimpleTableMetadata(),
+                 .storage_credentials = {{.prefix = "s3",
+                                          .config = {{"s3.access-key-id", "AKIAtest"},
+                                                     {"s3.secret-access-key", "secret"},
+                                                     {"s3.region", "us-east-1"}}}}}}),
     [](const ::testing::TestParamInfo<LoadTableResultParam>& info) {
       return info.param.test_name;
     });
@@ -1145,7 +1160,18 @@ INSTANTIATE_TEST_SUITE_P(
             .json_str =
                 R"({"metadata":{"format-version":2,"table-uuid":"test-uuid-1234","location":"s3://bucket/test","last-sequence-number":0,"last-updated-ms":0,"last-column-id":1,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0,"properties":{}},"config":{"warehouse":"s3://bucket/warehouse"}})",
             .expected_model = {.metadata = MakeSimpleTableMetadata(),
-                               .config = {{"warehouse", "s3://bucket/warehouse"}}}}),
+                               .config = {{"warehouse", "s3://bucket/warehouse"}}}},
+        LoadTableResultDeserializeParam{
+            .test_name = "WithStorageCredentials",
+            .json_str =
+                R"({"metadata":{"format-version":2,"table-uuid":"test-uuid-1234","location":"s3://bucket/test","last-sequence-number":0,"last-updated-ms":0,"last-column-id":1,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0,"properties":{}},"storage-credentials":[{"prefix":"s3","config":{"s3.access-key-id":"AKIAtest","s3.secret-access-key":"secret","s3.session-token":"token","s3.region":"us-east-1"}}]})",
+            .expected_model =
+                {.metadata = MakeSimpleTableMetadata(),
+                 .storage_credentials = {{.prefix = "s3",
+                                          .config = {{"s3.access-key-id", "AKIAtest"},
+                                                     {"s3.secret-access-key", "secret"},
+                                                     {"s3.session-token", "token"},
+                                                     {"s3.region", "us-east-1"}}}}}}),
     [](const ::testing::TestParamInfo<LoadTableResultDeserializeParam>& info) {
       return info.param.test_name;
     });
@@ -1184,7 +1210,28 @@ INSTANTIATE_TEST_SUITE_P(
         LoadTableResultInvalidParam{
             .test_name = "InvalidMetadataContent",
             .invalid_json_str = R"({"metadata":{"format-version":"invalid"}})",
-            .expected_error_message = "type must be number, but is string"}),
+            .expected_error_message = "type must be number, but is string"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialsNotArray",
+            .invalid_json_str = LoadTableJsonWithCredentials(R"("oops")"),
+            .expected_error_message = "Cannot parse storage credentials from non-array"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialMissingPrefix",
+            .invalid_json_str = LoadTableJsonWithCredentials(R"([{"config":{"k":"v"}}])"),
+            .expected_error_message = "Missing 'prefix'"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialMissingConfig",
+            .invalid_json_str = LoadTableJsonWithCredentials(R"([{"prefix":"s3"}])"),
+            .expected_error_message = "Missing 'config'"},
+        LoadTableResultInvalidParam{.test_name = "StorageCredentialEmptyPrefix",
+                                    .invalid_json_str = LoadTableJsonWithCredentials(
+                                        R"([{"prefix":"","config":{"k":"v"}}])"),
+                                    .expected_error_message = "prefix must be non-empty"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialEmptyConfig",
+            .invalid_json_str =
+                LoadTableJsonWithCredentials(R"([{"prefix":"s3","config":{}}])"),
+            .expected_error_message = "config must be non-empty"}),
     [](const ::testing::TestParamInfo<LoadTableResultInvalidParam>& info) {
       return info.param.test_name;
     });
@@ -2565,4 +2612,18 @@ TEST(FetchPlanningResultResponseRoundtripTest, FailedWithError) {
   EXPECT_EQ(*result, *result2);
 }
 
+TEST(StorageCredentialValidateTest, RequiresPrefixAndConfig) {
+  EXPECT_THAT(
+      (StorageCredential{.prefix = "s3", .config = {{"s3.region", "us"}}}.Validate()),
+      IsOk());
+
+  auto empty_prefix = StorageCredential{.prefix = "", .config = {{"s3.region", "us"}}};
+  EXPECT_THAT(empty_prefix.Validate(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(empty_prefix.Validate(), HasErrorMessage("prefix must be non-empty"));
+
+  auto empty_config = StorageCredential{.prefix = "s3", .config = {}};
+  EXPECT_THAT(empty_config.Validate(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(empty_config.Validate(), HasErrorMessage("config must be non-empty"));
+}
+
 }  // namespace iceberg::rest
diff --git a/src/iceberg/util/location_util.h b/src/iceberg/util/location_util.h
index eb78dece3..cbeac1574 100644
--- a/src/iceberg/util/location_util.h
+++ b/src/iceberg/util/location_util.h
@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include <string>
 #include <string_view>
 
 #include "iceberg/iceberg_export.h"
@@ -37,6 +38,27 @@ class ICEBERG_EXPORT LocationUtil {
     }
     return path;
   }
+
+  /// \brief Rewrites S3-compatible schemes (s3a://, s3n://, oss://) to s3:// so
+  /// locations and credential prefixes can be prefix-matched uniformly.
+  static std::string CanonicalizeS3Scheme(std::string_view location) {
+    for (std::string_view scheme : {"s3a://", "s3n://", "oss://"}) {
+      if (location.starts_with(scheme)) {
+        return std::string("s3://").append(location.substr(scheme.size()));
+      }
+    }
+    return std::string(location);
+  }
+
+  /// \brief True if `prefix` matches `path` at a path boundary (equal, next char
+  /// '/', or a bare scheme), so `s3://bucket` does not match `s3://bucket-x`.
+  static bool PathHasPrefix(std::string_view path, std::string_view prefix) {
+    if (!path.starts_with(prefix)) {
+      return false;
+    }
+    return path.size() == prefix.size() || path[prefix.size()] == '/' ||
+           prefix.find("://") == std::string_view::npos;
+  }
 };
 
 }  // namespace iceberg