refactor: move credential properties to FileIO, remove io_properties from Table

Move vended credential properties from Table::io_properties() to
FileIO::properties(), matching Python's table.io.properties pattern.
This is the standard surface for engines to access vended credentials.

- Add properties() accessor to FileIO base class
- FileIORegistry::Load() automatically populates FileIO::properties_
- Remove io_properties from Table::Make, StagedTable::Make, and Refresh
- Remove ResolvedTableIO struct (no longer needed)
- Upgrade Docker fixture to MinIO + S3 warehouse with credential vending
- Rewrite integration tests to verify vended S3 credentials flow through
- Clean up credential_vending_test.cc (remove Table accessor tests)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Sreesh Maheshwar

src/iceberg/arrow/arrow_fs_file_io.cc

@@ -79,25 +79,6 @@ Status ArrowFileSystemFileIO::WriteFile(const std::string& file_location,
   return {};
 }
 
-Result<std::shared_ptr<::arrow::io::RandomAccessFile>>
-ArrowFileSystemFileIO::OpenInputFile(const std::string& file_location,
-                                     std::optional<size_t> length) {
-  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
-  ::arrow::fs::FileInfo file_info(path, ::arrow::fs::FileType::File);
-  if (length.has_value()) {
-    file_info.set_size(length.value());
-  }
-  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, arrow_fs_->OpenInputFile(file_info));
-  return file;
-}
-
-Result<std::shared_ptr<::arrow::io::OutputStream>>
-ArrowFileSystemFileIO::OpenOutputStream(const std::string& file_location) {
-  ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
-  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto file, arrow_fs_->OpenOutputStream(path));
-  return file;
-}
-
 /// \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));

src/iceberg/arrow/arrow_fs_file_io_internal.h

@@ -20,10 +20,8 @@
 #pragma once
 
 #include <memory>
-#include <optional>
 
 #include <arrow/filesystem/filesystem.h>
-#include <arrow/io/interfaces.h>
 
 #include "iceberg/file_io.h"
 #include "iceberg/iceberg_bundle_export.h"
@@ -54,27 +52,6 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
   /// \brief Delete a file at the given location.
   Status DeleteFile(const std::string& file_location) override;
 
-  /// \brief Open an input file for reading, resolving any URI scheme in the path.
-  ///
-  /// This method resolves URI schemes (e.g., "s3://bucket/key" -> "bucket/key")
-  /// before opening the file. Use this instead of calling fs()->OpenInputFile()
-  /// directly to ensure URI resolution is applied consistently.
-  ///
-  /// \param file_location The file location, which may contain a URI scheme.
-  /// \param length Optional file size hint for the Arrow filesystem.
-  Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenInputFile(
-      const std::string& file_location, std::optional<size_t> length = std::nullopt);
-
-  /// \brief Open an output stream for writing, resolving any URI scheme in the path.
-  ///
-  /// This method resolves URI schemes (e.g., "s3://bucket/key" -> "bucket/key")
-  /// before opening the stream. Use this instead of calling fs()->OpenOutputStream()
-  /// directly to ensure URI resolution is applied consistently.
-  ///
-  /// \param file_location The file location, which may contain a URI scheme.
-  Result<std::shared_ptr<::arrow::io::OutputStream>> OpenOutputStream(
-      const std::string& file_location);
-
   /// \brief Get the Arrow file system.
   const std::shared_ptr<::arrow::fs::FileSystem>& fs() const { return arrow_fs_; }
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -364,10 +364,10 @@ Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
   ICEBERG_ASSIGN_OR_RAISE(auto result,
                           CreateTableInternal(identifier, schema, spec, order, location,
                                               properties, /*stage_create=*/false));
-  ICEBERG_ASSIGN_OR_RAISE(auto resolved, ResolveTableFileIO(result));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, ResolveTableFileIO(result));
   return Table::Make(identifier, std::move(result.metadata),
-                     std::move(result.metadata_location), std::move(resolved.io),
-                     shared_from_this(), std::move(resolved.props));
+                     std::move(result.metadata_location), std::move(table_io),
+                     shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
@@ -409,12 +409,11 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
   ICEBERG_ASSIGN_OR_RAISE(auto result,
                           CreateTableInternal(identifier, schema, spec, order, location,
                                               properties, /*stage_create=*/true));
-  ICEBERG_ASSIGN_OR_RAISE(auto resolved, ResolveTableFileIO(result));
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto staged_table,
-      StagedTable::Make(identifier, std::move(result.metadata),
-                        std::move(result.metadata_location), std::move(resolved.io),
-                        shared_from_this(), std::move(resolved.props)));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, ResolveTableFileIO(result));
+  ICEBERG_ASSIGN_OR_RAISE(auto staged_table,
+                          StagedTable::Make(identifier, std::move(result.metadata),
+                                            std::move(result.metadata_location),
+                                            std::move(table_io), shared_from_this()));
   return Transaction::Make(std::move(staged_table), TransactionKind::kCreate);
 }
 
@@ -477,10 +476,10 @@ Result<std::string> RestCatalog::LoadTableInternal(
   return response.body();
 }
 
-Result<RestCatalog::ResolvedTableIO> RestCatalog::ResolveTableFileIO(
+Result<std::shared_ptr<FileIO>> RestCatalog::ResolveTableFileIO(
     const LoadTableResult& result) const {
   if (result.config.empty() && result.storage_credentials.empty()) {
-    return ResolvedTableIO{.io = file_io_, .props = {}};
+    return file_io_;
   }
 
   // Merge order: catalog props < table config < storage credentials (highest priority)
@@ -514,23 +513,23 @@ Result<RestCatalog::ResolvedTableIO> RestCatalog::ResolveTableFileIO(
     ICEBERG_ASSIGN_OR_RAISE(auto kind, DetectBuiltinFileIO(warehouse));
     io_impl = std::string(BuiltinFileIOName(kind));
   } else {
-    return ResolvedTableIO{.io = file_io_, .props = std::move(merged)};
+    // Cannot detect FileIO type; fall back to catalog-level FileIO.
+    return file_io_;
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto table_io, FileIORegistry::Load(io_impl, merged));
-  return ResolvedTableIO{.io = std::shared_ptr<FileIO>(std::move(table_io)),
-                         .props = std::move(merged)};
+  return std::shared_ptr<FileIO>(std::move(table_io));
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& identifier) {
   ICEBERG_ASSIGN_OR_RAISE(const auto body, LoadTableInternal(identifier));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
 
-  ICEBERG_ASSIGN_OR_RAISE(auto resolved, ResolveTableFileIO(load_result));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, ResolveTableFileIO(load_result));
   return Table::Make(identifier, std::move(load_result.metadata),
-                     std::move(load_result.metadata_location), std::move(resolved.io),
-                     shared_from_this(), std::move(resolved.props));
+                     std::move(load_result.metadata_location), std::move(table_io),
+                     shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
@@ -551,10 +550,10 @@ Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
-  ICEBERG_ASSIGN_OR_RAISE(auto resolved, ResolveTableFileIO(load_result));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, ResolveTableFileIO(load_result));
   return Table::Make(identifier, std::move(load_result.metadata),
-                     std::move(load_result.metadata_location), std::move(resolved.io),
-                     shared_from_this(), std::move(resolved.props));
+                     std::move(load_result.metadata_location), std::move(table_io),
+                     shared_from_this());
 }
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_catalog.h

@@ -109,11 +109,7 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
-  struct ResolvedTableIO {
-    std::shared_ptr<FileIO> io;
-    std::unordered_map<std::string, std::string> props;
-  };
-  Result<ResolvedTableIO> ResolveTableFileIO(const LoadTableResult& result) const;
+  Result<std::shared_ptr<FileIO>> ResolveTableFileIO(const LoadTableResult& result) const;
 
   Result<LoadTableResult> CreateTableInternal(
       const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,

src/iceberg/file_io.h

@@ -22,6 +22,7 @@
 #include <optional>
 #include <string>
 #include <string_view>
+#include <unordered_map>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
@@ -42,6 +43,14 @@ class ICEBERG_EXPORT FileIO {
   FileIO() = default;
   virtual ~FileIO() = default;
 
+  /// \brief Returns the configuration properties used to initialize this FileIO.
+  ///
+  /// Engines that need to configure their own storage access (e.g., for credential
+  /// vending) can read these properties to obtain the resolved credentials.
+  const std::unordered_map<std::string, std::string>& properties() const {
+    return properties_;
+  }
+
   /// \brief Read the content of the file at the given location.
   ///
   /// \param file_location The location of the file to read.
@@ -73,6 +82,10 @@ class ICEBERG_EXPORT FileIO {
   virtual Status DeleteFile(const std::string& file_location) {
     return NotImplemented("DeleteFile not implemented");
   }
+
+ private:
+  friend class FileIORegistry;
+  std::unordered_map<std::string, std::string> properties_;
 };
 
 }  // namespace iceberg

src/iceberg/file_io_registry.cc

@@ -20,6 +20,8 @@
 #include <mutex>
 #include <utility>
 
+#include "iceberg/util/macros.h"
+
 namespace iceberg {
 
 namespace {
@@ -55,7 +57,9 @@ Result<std::unique_ptr<FileIO>> FileIORegistry::Load(
     }
     factory = it->second;
   }
-  return factory(properties);
+  ICEBERG_ASSIGN_OR_RAISE(auto io, factory(properties));
+  io->properties_ = properties;
+  return io;
 }
 
 }  // namespace iceberg

src/iceberg/table.cc

@@ -47,11 +47,11 @@
 
 namespace iceberg {
 
-Result<std::shared_ptr<Table>> Table::Make(
-    TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
-    std::string metadata_location, std::shared_ptr<FileIO> io,
-    std::shared_ptr<Catalog> catalog,
-    std::unordered_map<std::string, std::string> io_properties) {
+Result<std::shared_ptr<Table>> Table::Make(TableIdentifier identifier,
+                                           std::shared_ptr<TableMetadata> metadata,
+                                           std::string metadata_location,
+                                           std::shared_ptr<FileIO> io,
+                                           std::shared_ptr<Catalog> catalog) {
   if (metadata == nullptr) [[unlikely]] {
     return InvalidArgument("Metadata cannot be null");
   }
@@ -66,36 +66,29 @@ Result<std::shared_ptr<Table>> Table::Make(
   }
   return std::shared_ptr<Table>(new Table(std::move(identifier), std::move(metadata),
                                           std::move(metadata_location), std::move(io),
-                                          std::move(catalog), std::move(io_properties)));
+                                          std::move(catalog)));
 }
 
 Table::~Table() = default;
 
 Table::Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
              std::string metadata_location, std::shared_ptr<FileIO> io,
-             std::shared_ptr<Catalog> catalog,
-             std::unordered_map<std::string, std::string> io_properties)
+             std::shared_ptr<Catalog> catalog)
     : identifier_(std::move(identifier)),
       metadata_(std::move(metadata)),
       metadata_location_(std::move(metadata_location)),
       io_(std::move(io)),
       catalog_(std::move(catalog)),
-      metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())),
-      io_properties_(std::move(io_properties)) {}
+      metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())) {}
 
 const std::string& Table::uuid() const { return metadata_->table_uuid; }
 
-const std::unordered_map<std::string, std::string>& Table::io_properties() const {
-  return io_properties_;
-}
-
 Status Table::Refresh() {
   ICEBERG_ASSIGN_OR_RAISE(auto refreshed_table, catalog_->LoadTable(identifier_));
   if (metadata_location_ != refreshed_table->metadata_file_location()) {
     metadata_ = std::move(refreshed_table->metadata_);
     metadata_location_ = std::string(refreshed_table->metadata_file_location());
     io_ = std::move(refreshed_table->io_);
-    io_properties_ = std::move(refreshed_table->io_properties_);
     metadata_cache_ = std::make_unique<TableMetadataCache>(metadata_.get());
   }
   return {};
@@ -243,8 +236,7 @@ Result<std::shared_ptr<SnapshotManager>> Table::NewSnapshotManager() {
 Result<std::shared_ptr<StagedTable>> StagedTable::Make(
     TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
     std::string metadata_location, std::shared_ptr<FileIO> io,
-    std::shared_ptr<Catalog> catalog,
-    std::unordered_map<std::string, std::string> io_properties) {
+    std::shared_ptr<Catalog> catalog) {
   if (metadata == nullptr) [[unlikely]] {
     return InvalidArgument("Metadata cannot be null");
   }
@@ -254,9 +246,9 @@ Result<std::shared_ptr<StagedTable>> StagedTable::Make(
   if (catalog == nullptr) [[unlikely]] {
     return InvalidArgument("Catalog cannot be null");
   }
-  return std::shared_ptr<StagedTable>(new StagedTable(
-      std::move(identifier), std::move(metadata), std::move(metadata_location),
-      std::move(io), std::move(catalog), std::move(io_properties)));
+  return std::shared_ptr<StagedTable>(
+      new StagedTable(std::move(identifier), std::move(metadata),
+                      std::move(metadata_location), std::move(io), std::move(catalog)));
 }
 
 StagedTable::~StagedTable() = default;
@@ -276,8 +268,8 @@ Result<std::shared_ptr<StaticTable>> StaticTable::Make(
   }
   return std::shared_ptr<StaticTable>(
       new StaticTable(std::move(identifier), std::move(metadata),
-                      std::move(metadata_location), std::move(io), /*catalog=*/nullptr,
-                      /*io_properties=*/{}));
+                      std::move(metadata_location), std::move(io),
+                      /*catalog=*/nullptr));
 }
 
 StaticTable::~StaticTable() = default;

src/iceberg/table.h

@@ -43,13 +43,11 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \param[in] metadata_location The location of the table metadata file.
   /// \param[in] io The FileIO to read and write table data and metadata files.
   /// \param[in] catalog The catalog that this table belongs to.
-  /// \param[in] io_properties Per-table FileIO configuration properties provided by the
-  /// catalog.
-  static Result<std::shared_ptr<Table>> Make(
-      TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
-      std::string metadata_location, std::shared_ptr<FileIO> io,
-      std::shared_ptr<Catalog> catalog,
-      std::unordered_map<std::string, std::string> io_properties = {});
+  static Result<std::shared_ptr<Table>> Make(TableIdentifier identifier,
+                                             std::shared_ptr<TableMetadata> metadata,
+                                             std::string metadata_location,
+                                             std::shared_ptr<FileIO> io,
+                                             std::shared_ptr<Catalog> catalog);
 
   virtual ~Table();
 
@@ -113,9 +111,6 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Returns a FileIO to read and write table data and metadata files
   const std::shared_ptr<FileIO>& io() const;
 
-  /// \brief Returns per-table FileIO configuration properties provided by the catalog.
-  const std::unordered_map<std::string, std::string>& io_properties() const;
-
   /// \brief Returns the current metadata for this table
   const std::shared_ptr<TableMetadata>& metadata() const;
 
@@ -187,16 +182,14 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
  protected:
   Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
         std::string metadata_location, std::shared_ptr<FileIO> io,
-        std::shared_ptr<Catalog> catalog,
-        std::unordered_map<std::string, std::string> io_properties);
+        std::shared_ptr<Catalog> catalog);
 
   const TableIdentifier identifier_;
   std::shared_ptr<TableMetadata> metadata_;
   std::string metadata_location_;
   std::shared_ptr<FileIO> io_;
   std::shared_ptr<Catalog> catalog_;
   std::unique_ptr<class TableMetadataCache> metadata_cache_;
-  std::unordered_map<std::string, std::string> io_properties_;
 };
 
 /// \brief A table created by stage-create and not yet committed.
@@ -205,8 +198,7 @@ class ICEBERG_EXPORT StagedTable final : public Table {
   static Result<std::shared_ptr<StagedTable>> Make(
       TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
       std::string metadata_location, std::shared_ptr<FileIO> io,
-      std::shared_ptr<Catalog> catalog,
-      std::unordered_map<std::string, std::string> io_properties = {});
+      std::shared_ptr<Catalog> catalog);
 
   ~StagedTable() override;