feat(io): route FileIO through configured runtime

crates/catalog/glue/src/catalog.rs

@@ -221,6 +221,7 @@ impl GlueCatalog {
         });
         let file_io = FileIOBuilder::new(factory)
             .with_props(file_io_props)
+            .with_runtime(runtime.clone())
             .build();
 
         Ok(GlueCatalog {

crates/catalog/hms/src/catalog.rs

@@ -231,6 +231,7 @@ impl HmsCatalog {
         })?;
         let file_io = FileIOBuilder::new(factory)
             .with_props(&config.props)
+            .with_runtime(runtime.clone())
             .build();
 
         Ok(Self {

crates/catalog/rest/src/catalog.rs

@@ -483,7 +483,10 @@ impl RestCatalog {
                 )
             })?;
 
-        let file_io = FileIOBuilder::new(factory).with_props(props).build();
+        let file_io = FileIOBuilder::new(factory)
+            .with_props(props)
+            .with_runtime(self.runtime.clone())
+            .build();
 
         Ok(file_io)
     }

crates/catalog/s3tables/src/catalog.rs

@@ -220,6 +220,7 @@ impl S3TablesCatalog {
         });
         let file_io = FileIOBuilder::new(factory)
             .with_props(&config.props)
+            .with_runtime(runtime.clone())
             .build();
 
         Ok(Self {

crates/catalog/sql/src/catalog.rs

@@ -261,6 +261,7 @@ impl SqlCatalog {
         // Unrecognized keys are ignored by backends.
         let fileio = FileIOBuilder::new(factory)
             .with_props(config.props.clone())
+            .with_runtime(runtime.clone())
             .build();
 
         install_default_drivers();

crates/iceberg/public-api.txt

@@ -692,6 +692,7 @@ pub fn iceberg::io::FileIO::new_input(&self, path: impl core::convert::AsRef<str
 pub fn iceberg::io::FileIO::new_output(&self, path: impl core::convert::AsRef<str>) -> iceberg::Result<iceberg::io::OutputFile>
 pub fn iceberg::io::FileIO::new_with_fs() -> Self
 pub fn iceberg::io::FileIO::new_with_memory() -> Self
+pub fn iceberg::io::FileIO::with_runtime(self, runtime: iceberg::Runtime) -> Self
 impl core::clone::Clone for iceberg::io::FileIO
 pub fn iceberg::io::FileIO::clone(&self) -> iceberg::io::FileIO
 impl core::fmt::Debug for iceberg::io::FileIO
@@ -703,6 +704,7 @@ pub fn iceberg::io::FileIOBuilder::config(&self) -> &iceberg::io::StorageConfig
 pub fn iceberg::io::FileIOBuilder::new(factory: alloc::sync::Arc<dyn iceberg::io::StorageFactory>) -> Self
 pub fn iceberg::io::FileIOBuilder::with_prop(self, key: impl alloc::string::ToString, value: impl alloc::string::ToString) -> Self
 pub fn iceberg::io::FileIOBuilder::with_props(self, args: impl core::iter::traits::collect::IntoIterator<Item = (impl alloc::string::ToString, impl alloc::string::ToString)>) -> Self
+pub fn iceberg::io::FileIOBuilder::with_runtime(self, runtime: iceberg::Runtime) -> Self
 impl core::clone::Clone for iceberg::io::FileIOBuilder
 pub fn iceberg::io::FileIOBuilder::clone(&self) -> iceberg::io::FileIOBuilder
 impl core::fmt::Debug for iceberg::io::FileIOBuilder
@@ -3029,6 +3031,7 @@ pub fn iceberg::table::Table::metadata_ref(&self) -> iceberg::spec::TableMetadat
 pub fn iceberg::table::Table::reader_builder(&self) -> iceberg::arrow::ArrowReaderBuilder
 pub fn iceberg::table::Table::readonly(&self) -> bool
 pub fn iceberg::table::Table::scan(&self) -> iceberg::scan::TableScanBuilder<'_>
+pub fn iceberg::table::Table::with_runtime(self, runtime: iceberg::Runtime) -> Self
 impl core::clone::Clone for iceberg::table::Table
 pub fn iceberg::table::Table::clone(&self) -> iceberg::table::Table
 impl core::fmt::Debug for iceberg::table::Table

crates/iceberg/src/catalog/memory/catalog.rs

@@ -145,7 +145,10 @@ impl MemoryCatalog {
 
         Ok(Self {
             root_namespace_state: Mutex::new(NamespaceState::default()),
-            file_io: FileIOBuilder::new(factory).with_props(config.props).build(),
+            file_io: FileIOBuilder::new(factory)
+                .with_props(config.props)
+                .with_runtime(runtime.clone())
+                .build(),
             warehouse_location: config.warehouse,
             runtime,
         })