feat(rest): support vended storage credentials with per-prefix FileIO

crates/catalog/rest/src/catalog.rs

@@ -45,7 +45,7 @@ use crate::endpoint::{Endpoint, V1_NAMESPACE_EXISTS, V1_TABLE_EXISTS};
 use crate::types::{
     CatalogConfig, CommitTableRequest, CommitTableResponse, CreateNamespaceRequest,
     CreateTableRequest, ListNamespaceResponse, ListTablesResponse, LoadTableResult,
-    NamespaceResponse, RegisterTableRequest, RenameTableRequest,
+    NamespaceResponse, RegisterTableRequest, RenameTableRequest, StorageCredential,
 };
 
 /// REST catalog URI
@@ -509,6 +509,7 @@ impl RestCatalog {
         &self,
         metadata_location: Option<&str>,
         extra_config: Option<HashMap<String, String>>,
+        storage_credentials: Option<&[StorageCredential]>,
     ) -> Result<FileIO> {
         let mut props = self.context().await?.config.props.clone();
         if let Some(config) = extra_config {
@@ -541,9 +542,20 @@ impl RestCatalog {
                 )
             })?;
 
-        let file_io = FileIOBuilder::new(factory).with_props(props).build();
+        let mut builder = FileIOBuilder::new(factory).with_props(props.clone());
 
-        Ok(file_io)
+        // Vended credentials are scoped per location prefix: give each its own
+        // storage. Paths under no vended prefix fall back to the default `props`
+        // above, which carry no credentials.
+        if let Some(creds) = storage_credentials {
+            for cred in creds {
+                let mut prefixed = props.clone();
+                prefixed.extend(cred.config.clone());
+                builder = builder.with_prefixed_props(cred.prefix.clone(), prefixed);
+            }
+        }
+
+        Ok(builder.build())
     }
 
     /// Invalidate the current token without generating a new one. On the next request, the client
@@ -846,14 +858,15 @@ impl Catalog for RestCatalog {
             "Metadata location missing in `create_table` response!",
         ))?;
 
-        let config = response
-            .config
-            .into_iter()
-            .chain(self.user_config.props.clone())
-            .collect();
+        let mut base_config = response.config.clone();
+        base_config.extend(self.user_config.props.clone());
 
         let file_io = self
-            .load_file_io(Some(metadata_location), Some(config))
+            .load_file_io(
+                Some(metadata_location),
+                Some(base_config),
+                response.storage_credentials.as_deref(),
+            )
             .await?;
 
         let mut table_builder = Table::builder()
@@ -880,6 +893,9 @@ impl Catalog for RestCatalog {
     async fn load_table(&self, table_ident: &TableIdent) -> Result<Table> {
         let context = self.context().await?;
 
+        // Vended credentials are opt-in via a `header.X-Iceberg-Access-Delegation`
+        // catalog property (applied to every request like the Iceberg Java client);
+        // any returned `storage_credentials` are wired into the FileIO below.
         let request = context
             .client
             .request(Method::GET, context.config.table_endpoint(table_ident))
@@ -906,14 +922,15 @@ impl Catalog for RestCatalog {
             }
         };
 
-        let config = response
-            .config
-            .into_iter()
-            .chain(self.user_config.props.clone())
-            .collect();
+        let mut base_config = response.config.clone();
+        base_config.extend(self.user_config.props.clone());
 
         let file_io = self
-            .load_file_io(response.metadata_location.as_deref(), Some(config))
+            .load_file_io(
+                response.metadata_location.as_deref(),
+                Some(base_config),
+                response.storage_credentials.as_deref(),
+            )
             .await?;
 
         let mut table_builder = Table::builder()
@@ -1048,7 +1065,9 @@ 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(Some(metadata_location), None, None)
+            .await?;
 
         let mut table_builder = Table::builder()
             .identifier(table_ident.clone())
@@ -1122,8 +1141,10 @@ impl Catalog for RestCatalog {
             }
         };
 
+        // The commit response carries no credentials, so build a plain FileIO;
+        // the transaction layer reuses the credentialed one it already holds.
         let file_io = self
-            .load_file_io(Some(&response.metadata_location), None)
+            .load_file_io(Some(&response.metadata_location), None, None)
             .await?;
 
         let mut table_builder = Table::builder()
@@ -2671,6 +2692,90 @@ mod tests {
         rename_table_mock.assert_async().await;
     }
 
+    #[tokio::test]
+    async fn test_load_table_uses_vended_credentials() {
+        let mut server = Server::new_async().await;
+
+        let config_mock = create_config_mock(&mut server).await;
+
+        // Vended credentials are opt-in via a `header.*` catalog property (like the
+        // Java client). With it configured, the header is sent and the response's
+        // `storage-credentials` are accepted (the FileIO builds).
+        let load_table_mock = server
+            .mock("GET", "/v1/namespaces/ns1/tables/test1")
+            .match_header("x-iceberg-access-delegation", "vended-credentials")
+            .with_status(200)
+            .with_body_from_file(format!(
+                "{}/testdata/{}",
+                env!("CARGO_MANIFEST_DIR"),
+                "load_table_response_with_credentials.json"
+            ))
+            .create_async()
+            .await;
+
+        let props = HashMap::from([(
+            "header.X-Iceberg-Access-Delegation".to_string(),
+            "vended-credentials".to_string(),
+        )]);
+        let catalog = RestCatalog::new(
+            RestCatalogConfig::builder()
+                .uri(server.url())
+                .props(props)
+                .build(),
+            Some(Arc::new(LocalFsStorageFactory)),
+            Runtime::current(),
+            None,
+        );
+
+        let table = catalog
+            .load_table(&TableIdent::from_strs(["ns1", "test1"]).unwrap())
+            .await
+            .unwrap();
+
+        assert_eq!(
+            "s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json",
+            table.metadata_location().unwrap()
+        );
+
+        config_mock.assert_async().await;
+        load_table_mock.assert_async().await;
+    }
+
+    #[tokio::test]
+    async fn test_load_table_omits_delegation_header_by_default() {
+        let mut server = Server::new_async().await;
+
+        let config_mock = create_config_mock(&mut server).await;
+
+        // No delegation header is hardcoded: without a `header.*` prop, none is sent.
+        let load_table_mock = server
+            .mock("GET", "/v1/namespaces/ns1/tables/test1")
+            .match_header("x-iceberg-access-delegation", mockito::Matcher::Missing)
+            .with_status(200)
+            .with_body_from_file(format!(
+                "{}/testdata/{}",
+                env!("CARGO_MANIFEST_DIR"),
+                "load_table_response.json"
+            ))
+            .create_async()
+            .await;
+
+        let catalog = RestCatalog::new(
+            RestCatalogConfig::builder().uri(server.url()).build(),
+            Some(Arc::new(LocalFsStorageFactory)),
+            Runtime::current(),
+            None,
+        );
+
+        catalog
+            .load_table(&TableIdent::from_strs(["ns1", "test1"]).unwrap())
+            .await
+            .unwrap();
+
+        config_mock.assert_async().await;
+        load_table_mock.assert_async().await;
+    }
+
     #[tokio::test]
     async fn test_create_table() {
         let mut server = Server::new_async().await;
@@ -2888,6 +2993,7 @@ mod tests {
 
         let config_mock = create_config_mock(&mut server).await;
 
+        // GET hit once: the transaction refreshes the table before committing.
         let load_table_mock = server
             .mock("GET", "/v1/namespaces/ns1/tables/test1")
             .with_status(200)
@@ -2896,6 +3002,7 @@ mod tests {
                 env!("CARGO_MANIFEST_DIR"),
                 "load_table_response.json"
             ))
+            .expect(1)
             .create_async()
             .await;
 

crates/catalog/rest/testdata/load_table_response_with_credentials.json

@@ -0,0 +1,78 @@
+{
+  "metadata-location": "s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json",
+  "metadata": {
+    "format-version": 1,
+    "table-uuid": "b55d9dda-6561-423a-8bfc-787980ce421f",
+    "location": "s3://warehouse/database/table",
+    "last-updated-ms": 1646787054459,
+    "last-column-id": 2,
+    "schema": {
+      "type": "struct",
+      "schema-id": 0,
+      "fields": [
+        {"id": 1, "name": "id", "required": false, "type": "int"},
+        {"id": 2, "name": "data", "required": false, "type": "string"}
+      ]
+    },
+    "current-schema-id": 0,
+    "schemas": [
+      {
+        "type": "struct",
+        "schema-id": 0,
+        "fields": [
+          {"id": 1, "name": "id", "required": false, "type": "int"},
+          {"id": 2, "name": "data", "required": false, "type": "string"}
+        ]
+      }
+    ],
+    "partition-spec": [],
+    "default-spec-id": 0,
+    "partition-specs": [{"spec-id": 0, "fields": []}],
+    "last-partition-id": 999,
+    "default-sort-order-id": 0,
+    "sort-orders": [{"order-id": 0, "fields": []}],
+    "properties": {"owner": "bryan", "write.metadata.compression-codec": "gzip"},
+    "current-snapshot-id": 3497810964824022504,
+    "refs": {"main": {"snapshot-id": 3497810964824022504, "type": "branch"}},
+    "snapshots": [
+      {
+        "snapshot-id": 3497810964824022504,
+        "timestamp-ms": 1646787054459,
+        "summary": {
+          "operation": "append",
+          "spark.app.id": "local-1646787004168",
+          "added-data-files": "1",
+          "added-records": "1",
+          "added-files-size": "697",
+          "changed-partition-count": "1",
+          "total-records": "1",
+          "total-files-size": "697",
+          "total-data-files": "1",
+          "total-delete-files": "0",
+          "total-position-deletes": "0",
+          "total-equality-deletes": "0"
+        },
+        "manifest-list": "s3://warehouse/database/table/metadata/snap-3497810964824022504-1-c4f68204-666b-4e50-a9df-b10c34bf6b82.avro",
+        "schema-id": 0
+      }
+    ],
+    "snapshot-log": [{"timestamp-ms": 1646787054459, "snapshot-id": 3497810964824022504}],
+    "metadata-log": [
+      {
+        "timestamp-ms": 1646787031514,
+        "metadata-file": "s3://warehouse/database/table/metadata/00000-88484a1c-00e5-4a07-a787-c0e7aeffa805.gz.metadata.json"
+      }
+    ]
+  },
+  "config": {"client.factory": "io.tabular.iceberg.catalog.TabularAwsClientFactory", "region": "us-west-2"},
+  "storage-credentials": [
+    {
+      "prefix": "s3://warehouse/database/table",
+      "config": {
+        "s3.access-key-id": "vended-key-id",
+        "s3.secret-access-key": "vended-secret",
+        "s3.session-token": "vended-token"
+      }
+    }
+  ]
+}

crates/iceberg/public-api.txt

@@ -739,6 +739,7 @@ impl iceberg::io::FileIOBuilder
 pub fn iceberg::io::FileIOBuilder::build(self) -> iceberg::io::FileIO
 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_prefixed_props(self, prefix: impl core::convert::Into<alloc::string::String>, props: impl core::iter::traits::collect::IntoIterator<Item = (impl alloc::string::ToString, impl alloc::string::ToString)>) -> 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
 impl core::clone::Clone for iceberg::io::FileIOBuilder