feat!(storage): Allow StorageFactory::build to take table metadata

bindings/python/src/datafusion_table_provider.rs

@@ -118,7 +118,9 @@ impl PyIcebergDataFusionTable {
                 builder = builder.with_props(props);
             }
 
-            let file_io = builder.build();
+            let file_io = builder
+                .build(None)
+                .map_err(|e| PyRuntimeError::new_err(format!("Failed to build FileIO: {e}")))?;
 
             let static_table =
                 StaticTable::from_metadata_file(&metadata_location, table_ident, file_io)

crates/catalog/glue/src/catalog.rs

@@ -209,7 +209,7 @@ impl GlueCatalog {
         });
         let file_io = FileIOBuilder::new(factory)
             .with_props(file_io_props)
-            .build();
+            .build(None)?;
 
         Ok(GlueCatalog {
             config,

crates/catalog/hms/src/catalog.rs

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

crates/catalog/hms/tests/hms_catalog_test.rs

@@ -64,7 +64,8 @@ async fn get_catalog() -> HmsCatalog {
         customized_credential_load: None,
     }))
     .with_props(props.clone())
-    .build();
+    .build(None)
+    .unwrap();
 
     let mut retries = 0;
     while retries < 30 {

crates/catalog/loader/tests/common/mod.rs

@@ -236,7 +236,8 @@ async fn glue_catalog() -> GlueCatalog {
         customized_credential_load: None,
     }))
     .with_props(props.clone())
-    .build();
+    .build(None)
+    .unwrap();
 
     let mut retries = 0;
     while retries < 30 {
@@ -287,7 +288,8 @@ async fn hms_catalog() -> HmsCatalog {
         customized_credential_load: None,
     }))
     .with_props(props.clone())
-    .build();
+    .build(None)
+    .unwrap();
 
     let mut retries = 0;
     while retries < 30 {

crates/catalog/rest/src/catalog.rs

@@ -24,6 +24,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use iceberg::io::{FileIO, FileIOBuilder, StorageFactory};
+use iceberg::spec::TableMetadata;
 use iceberg::table::Table;
 use iceberg::{
     Catalog, CatalogBuilder, Error, ErrorKind, Namespace, NamespaceIdent, Result, TableCommit,
@@ -33,7 +34,7 @@ use itertools::Itertools;
 use reqwest::header::{
     HeaderMap, HeaderName, HeaderValue, {self},
 };
-use reqwest::{Client, Method, StatusCode, Url};
+use reqwest::{Client, Method, StatusCode};
 use tokio::sync::OnceCell;
 use typed_builder::TypedBuilder;
 
@@ -406,29 +407,14 @@ impl RestCatalog {
 
     async fn load_file_io(
         &self,
-        metadata_location: Option<&str>,
+        metadata: &TableMetadata,
         extra_config: Option<HashMap<String, String>>,
     ) -> Result<FileIO> {
         let mut props = self.context().await?.config.props.clone();
         if let Some(config) = extra_config {
             props.extend(config);
         }
 
-        // If the warehouse is a logical identifier instead of a URL we don't want
-        // to raise an exception
-        let warehouse_path = match self.context().await?.config.warehouse.as_deref() {
-            Some(url) if Url::parse(url).is_ok() => Some(url),
-            Some(_) => None,
-            None => None,
-        };
-
-        if metadata_location.or(warehouse_path).is_none() {
-            return Err(Error::new(
-                ErrorKind::Unexpected,
-                "Unable to load file io, neither warehouse nor metadata location is set!",
-            ));
-        }
-
         // Require a StorageFactory to be provided
         let factory = self
             .storage_factory
@@ -440,7 +426,9 @@ impl RestCatalog {
                 )
             })?;
 
-        let file_io = FileIOBuilder::new(factory).with_props(props).build();
+        let file_io = FileIOBuilder::new(factory)
+            .with_props(props)
+            .build(Some(metadata))?;
 
         Ok(file_io)
     }
@@ -743,20 +731,20 @@ impl Catalog for RestCatalog {
             }
         };
 
-        let metadata_location = response.metadata_location.as_ref().ok_or(Error::new(
-            ErrorKind::DataInvalid,
-            "Metadata location missing in `create_table` response!",
-        ))?;
+        if response.metadata_location.is_none() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "Metadata location missing in `create_table` response!",
+            ));
+        }
 
         let config = response
             .config
             .into_iter()
             .chain(self.user_config.props.clone())
             .collect();
 
-        let file_io = self
-            .load_file_io(Some(metadata_location), Some(config))
-            .await?;
+        let file_io = self.load_file_io(&response.metadata, Some(config)).await?;
 
         let table_builder = Table::builder()
             .identifier(table_ident.clone())
@@ -810,9 +798,7 @@ impl Catalog for RestCatalog {
             .chain(self.user_config.props.clone())
             .collect();
 
-        let file_io = self
-            .load_file_io(response.metadata_location.as_deref(), Some(config))
-            .await?;
+        let file_io = self.load_file_io(&response.metadata, Some(config)).await?;
 
         let table_builder = Table::builder()
             .identifier(table_ident.clone())
@@ -960,7 +946,7 @@ impl Catalog for RestCatalog {
             "Metadata location missing in `register_table` response!",
         ))?;
 
-        let file_io = self.load_file_io(Some(metadata_location), None).await?;
+        let file_io = self.load_file_io(&response.metadata, None).await?;
 
         Table::builder()
             .identifier(table_ident.clone())
@@ -1030,9 +1016,7 @@ impl Catalog for RestCatalog {
             }
         };
 
-        let file_io = self
-            .load_file_io(Some(&response.metadata_location), None)
-            .await?;
+        let file_io = self.load_file_io(&response.metadata, None).await?;
 
         Table::builder()
             .identifier(commit.identifier().clone())

crates/catalog/s3tables/src/catalog.rs

@@ -208,7 +208,7 @@ impl S3TablesCatalog {
         });
         let file_io = FileIOBuilder::new(factory)
             .with_props(&config.props)
-            .build();
+            .build(None)?;
 
         Ok(Self {
             config,

crates/catalog/sql/src/catalog.rs

@@ -244,7 +244,7 @@ impl SqlCatalog {
                 "StorageFactory must be provided for SqlCatalog. Use `with_storage_factory` to configure it.",
             )
         })?;
-        let fileio = FileIOBuilder::new(factory).build();
+        let fileio = FileIOBuilder::new(factory).build(None)?;
 
         install_default_drivers();
         let max_connections: u32 = config

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

@@ -134,7 +134,9 @@ 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)
+                .build(None)?,
             warehouse_location: config.warehouse,
         })
     }