feat(rest): support atomic multi-table commits

crates/catalog/rest/public-api.txt

@@ -57,6 +57,19 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::CommitTableResponse
 pub fn iceberg_catalog_rest::CommitTableResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
 impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::CommitTableResponse
 pub fn iceberg_catalog_rest::CommitTableResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+pub struct iceberg_catalog_rest::CommitTransactionRequest
+pub iceberg_catalog_rest::CommitTransactionRequest::table_changes: alloc::vec::Vec<iceberg_catalog_rest::CommitTableRequest>
+impl core::clone::Clone for iceberg_catalog_rest::CommitTransactionRequest
+pub fn iceberg_catalog_rest::CommitTransactionRequest::clone(&self) -> iceberg_catalog_rest::CommitTransactionRequest
+impl core::cmp::PartialEq for iceberg_catalog_rest::CommitTransactionRequest
+pub fn iceberg_catalog_rest::CommitTransactionRequest::eq(&self, other: &iceberg_catalog_rest::CommitTransactionRequest) -> bool
+impl core::fmt::Debug for iceberg_catalog_rest::CommitTransactionRequest
+pub fn iceberg_catalog_rest::CommitTransactionRequest::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl core::marker::StructuralPartialEq for iceberg_catalog_rest::CommitTransactionRequest
+impl serde_core::ser::Serialize for iceberg_catalog_rest::CommitTransactionRequest
+pub fn iceberg_catalog_rest::CommitTransactionRequest::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
+impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::CommitTransactionRequest
+pub fn iceberg_catalog_rest::CommitTransactionRequest::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub struct iceberg_catalog_rest::CreateNamespaceRequest
 pub iceberg_catalog_rest::CreateNamespaceRequest::namespace: iceberg::catalog::NamespaceIdent
 pub iceberg_catalog_rest::CreateNamespaceRequest::properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>
@@ -280,6 +293,8 @@ pub fn iceberg_catalog_rest::RenameTableRequest::serialize<__S>(&self, __seriali
 impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::RenameTableRequest
 pub fn iceberg_catalog_rest::RenameTableRequest::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub struct iceberg_catalog_rest::RestCatalog
+impl iceberg_catalog_rest::RestCatalog
+pub async fn iceberg_catalog_rest::RestCatalog::commit_transaction(&self, commits: alloc::vec::Vec<iceberg::catalog::TableCommit>) -> iceberg::error::Result<()>
 impl core::fmt::Debug for iceberg_catalog_rest::RestCatalog
 pub fn iceberg_catalog_rest::RestCatalog::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 impl iceberg::catalog::Catalog for iceberg_catalog_rest::RestCatalog
@@ -314,6 +329,8 @@ pub fn iceberg_catalog_rest::RestCatalogBuilder::with_kms_client_factory(self, k
 pub fn iceberg_catalog_rest::RestCatalogBuilder::with_runtime(self, runtime: iceberg::runtime::Runtime) -> Self
 pub fn iceberg_catalog_rest::RestCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc<dyn iceberg::io::storage::StorageFactory>) -> Self
 pub struct iceberg_catalog_rest::RestSessionCatalog
+impl iceberg_catalog_rest::RestSessionCatalog
+pub async fn iceberg_catalog_rest::RestSessionCatalog::commit_transaction(&self, context: &iceberg::catalog::session::SessionContext, commits: alloc::vec::Vec<iceberg::catalog::TableCommit>) -> iceberg::error::Result<()>
 impl core::fmt::Debug for iceberg_catalog_rest::RestSessionCatalog
 pub fn iceberg_catalog_rest::RestSessionCatalog::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 impl iceberg::catalog::session::SessionCatalog for iceberg_catalog_rest::RestSessionCatalog

crates/catalog/rest/src/catalog.rs

@@ -45,13 +45,13 @@ use crate::auth::{
 use crate::client::{
     HttpClient, deserialize_catalog_response, deserialize_unexpected_catalog_error,
 };
-use crate::endpoint::{Endpoint, V1_NAMESPACE_EXISTS, V1_TABLE_EXISTS};
+use crate::endpoint::{Endpoint, V1_COMMIT_TRANSACTION, V1_NAMESPACE_EXISTS, V1_TABLE_EXISTS};
 use crate::request::HttpRequest;
 use crate::response::HttpResponse;
 use crate::types::{
-    CatalogConfig, CommitTableRequest, CommitTableResponse, CreateNamespaceRequest,
-    CreateTableRequest, ListNamespaceResponse, ListTablesResponse, LoadTableResult,
-    NamespaceResponse, RegisterTableRequest, RenameTableRequest,
+    CatalogConfig, CommitTableRequest, CommitTableResponse, CommitTransactionRequest,
+    CreateNamespaceRequest, CreateTableRequest, ListNamespaceResponse, ListTablesResponse,
+    LoadTableResult, NamespaceResponse, RegisterTableRequest, RenameTableRequest,
 };
 
 /// REST catalog URI
@@ -241,6 +241,10 @@ impl RestCatalogConfig {
         self.url_prefixed(&["tables", "rename"])
     }
 
+    fn commit_transaction_endpoint(&self) -> String {
+        self.url_prefixed(&["transactions", "commit"])
+    }
+
     fn register_table_endpoint(&self, ns: &NamespaceIdent) -> String {
         self.url_prefixed(&["namespaces", &ns.to_url_string(), "register"])
     }
@@ -599,6 +603,14 @@ impl RestCatalog {
     async fn client(&self) -> Result<&RestCatalogClient> {
         self.inner.client().await
     }
+
+    /// Commits changes to multiple tables atomically; see
+    /// [`RestSessionCatalog::commit_transaction`].
+    pub async fn commit_transaction(&self, commits: Vec<TableCommit>) -> Result<()> {
+        self.inner
+            .commit_transaction(&self.session_context, commits)
+            .await
+    }
 }
 
 /// Every operation forwards to its [`RestSessionCatalog`] equivalent with the
@@ -843,6 +855,70 @@ impl RestSessionCatalog {
             .await
     }
 
+    /// Commits changes to multiple tables atomically through
+    /// `POST /v1/{prefix}/transactions/commit`: every table is updated, or none is.
+    ///
+    /// Build each commit with
+    /// [`Transaction::prepare_commit`](iceberg::transaction::Transaction::prepare_commit).
+    /// The server returns no metadata, so reload the tables afterwards. On a
+    /// conflict, reload the tables and rebuild their transactions before retrying.
+    pub async fn commit_transaction(
+        &self,
+        context: &SessionContext,
+        commits: Vec<TableCommit>,
+    ) -> Result<()> {
+        if commits.is_empty() {
+            return Ok(());
+        }
+        if !self.supports_endpoint(&V1_COMMIT_TRANSACTION).await? {
+            return Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "The REST server does not support multi-table commits",
+            ));
+        }
+
+        let client = self.client().await?;
+        let table_changes = commits
+            .into_iter()
+            .map(|mut commit| CommitTableRequest {
+                identifier: Some(commit.identifier().clone()),
+                requirements: commit.take_requirements(),
+                updates: commit.take_updates(),
+            })
+            .collect();
+        let request = HttpRequest::build(
+            client
+                .http_client
+                .request(Method::POST, client.config.commit_transaction_endpoint())
+                .json(&CommitTransactionRequest { table_changes }),
+        )?;
+
+        let http_response = client.query_catalog(context, request).await?;
+
+        match http_response.status() {
+            StatusCode::NO_CONTENT | StatusCode::OK => Ok(()),
+            StatusCode::NOT_FOUND => Err(Error::new(
+                ErrorKind::TableNotFound,
+                "Tried to update a table that does not exist",
+            )),
+            StatusCode::CONFLICT => Err(Error::new(
+                ErrorKind::CatalogCommitConflicts,
+                "CatalogCommitConflicts, one or more requirements failed. The client may retry.",
+            )
+            .with_retryable(true)),
+            StatusCode::INTERNAL_SERVER_ERROR
+            | StatusCode::BAD_GATEWAY
+            | StatusCode::GATEWAY_TIMEOUT => Err(Error::new(
+                ErrorKind::Unexpected,
+                "A server-side problem occurred; the commit state is unknown.",
+            )),
+            _ => Err(deserialize_unexpected_catalog_error(
+                http_response,
+                client.http_client.disable_header_redaction(),
+            )),
+        }
+    }
+
     /// Returns whether the server supports `endpoint`, per the `endpoints` it
     /// advertised in `GET /v1/config` (or a default base set when it advertised
     /// none).
@@ -4412,6 +4488,106 @@ mod tests {
         load_table_mock.assert_async().await
     }
 
+    /// A commit setting one property on `ns1.{name}`.
+    async fn property_commit(name: &str) -> TableCommit {
+        let file = File::open(format!(
+            "{}/testdata/create_table_response.json",
+            env!("CARGO_MANIFEST_DIR")
+        ))
+        .unwrap();
+        let resp = serde_json::from_reader::<_, LoadTableResult>(BufReader::new(file)).unwrap();
+        let table = Table::builder()
+            .metadata(resp.metadata)
+            .metadata_location(resp.metadata_location.unwrap())
+            .identifier(TableIdent::from_strs(["ns1", name]).unwrap())
+            .file_io(FileIO::new_with_fs())
+            .runtime(test_runtime())
+            .build()
+            .unwrap();
+        let tx = Transaction::new(&table);
+        let mut tx = tx
+            .update_table_properties()
+            .set("k".to_string(), "v".to_string())
+            .apply(tx)
+            .unwrap();
+        tx.prepare_commit().await.unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_commit_transaction() {
+        let mut server = Server::new_async().await;
+        let _config = create_config_mock(&mut server).await;
+        let commit_mock = server
+            .mock("POST", "/v1/transactions/commit")
+            .match_body(mockito::Matcher::AllOf(vec![
+                mockito::Matcher::Regex(r#""table-changes""#.to_string()),
+                mockito::Matcher::Regex(r#""name":"test1""#.to_string()),
+                mockito::Matcher::Regex(r#""name":"test2""#.to_string()),
+            ]))
+            .with_status(204)
+            .create_async()
+            .await;
+        let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build());
+
+        catalog
+            .commit_transaction(&SessionContext::empty(), vec![
+                property_commit("test1").await,
+                property_commit("test2").await,
+            ])
+            .await
+            .unwrap();
+
+        commit_mock.assert_async().await;
+    }
+
+    #[tokio::test]
+    async fn test_commit_transaction_conflict() {
+        let mut server = Server::new_async().await;
+        let _config = create_config_mock(&mut server).await;
+        let _commit = server
+            .mock("POST", "/v1/transactions/commit")
+            .with_status(409)
+            .create_async()
+            .await;
+        let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build());
+
+        let err = catalog
+            .commit_transaction(&SessionContext::empty(), vec![
+                property_commit("test1").await,
+            ])
+            .await
+            .unwrap_err();
+
+        assert_eq!(err.kind(), ErrorKind::CatalogCommitConflicts);
+    }
+
+    #[tokio::test]
+    async fn test_commit_transaction_unsupported() {
+        let mut server = Server::new_async().await;
+        let _config = create_config_mock_with_exists_endpoints(&mut server).await;
+        let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build());
+
+        let err = catalog
+            .commit_transaction(&SessionContext::empty(), vec![
+                property_commit("test1").await,
+            ])
+            .await
+            .unwrap_err();
+
+        assert_eq!(err.kind(), ErrorKind::FeatureUnsupported);
+    }
+
+    #[tokio::test]
+    async fn test_commit_transaction_empty() {
+        let server = Server::new_async().await;
+        let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build());
+
+        catalog
+            .commit_transaction(&SessionContext::empty(), vec![])
+            .await
+            .unwrap();
+    }
+
     #[tokio::test]
     async fn test_update_table_404() {
         let mut server = Server::new_async().await;

crates/catalog/rest/src/types.rs

@@ -294,6 +294,14 @@ pub struct CommitTableRequest {
     pub updates: Vec<TableUpdate>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "kebab-case")]
+/// Request to commit updates to multiple tables atomically.
+pub struct CommitTransactionRequest {
+    /// Changes for each table, applied all together or not at all
+    pub table_changes: Vec<CommitTableRequest>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "kebab-case")]
 /// Response returned when a table is successfully updated.

crates/iceberg/public-api.txt

@@ -3323,6 +3323,7 @@ pub async fn iceberg::transaction::Transaction::commit(self, catalog: &dyn icebe
 pub fn iceberg::transaction::Transaction::expire_snapshots(&self) -> iceberg::transaction::ExpireSnapshotsAction
 pub fn iceberg::transaction::Transaction::fast_append(&self) -> iceberg::transaction::FastAppendAction
 pub fn iceberg::transaction::Transaction::new(table: &iceberg::table::Table) -> Self
+pub async fn iceberg::transaction::Transaction::prepare_commit(&mut self) -> iceberg::Result<iceberg::TableCommit>
 pub fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::ReplaceSortOrderAction
 pub fn iceberg::transaction::Transaction::update_location(&self) -> iceberg::transaction::UpdateLocationAction
 pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::transaction::UpdateSchemaAction

crates/iceberg/src/transaction/mod.rs

@@ -217,6 +217,17 @@ impl Transaction {
             self.table = refreshed.clone();
         }
 
+        let table_commit = self.prepare_commit().await?;
+        catalog.update_table(table_commit).await
+    }
+
+    /// Builds the [`TableCommit`] for this transaction against its base table,
+    /// without refreshing the base or sending the commit.
+    ///
+    /// This lets several tables be committed atomically through a catalog that
+    /// supports multi-table commits. If that commit conflicts, reload the tables
+    /// and rebuild their transactions.
+    pub async fn prepare_commit(&mut self) -> Result<TableCommit> {
         let mut current_table = self.table.clone();
         let mut existing_updates: Vec<TableUpdate> = vec![];
         let mut existing_requirements: Vec<TableRequirement> = vec![];
@@ -232,13 +243,11 @@ impl Transaction {
             )?;
         }
 
-        let table_commit = TableCommit::builder()
+        Ok(TableCommit::builder()
             .ident(self.table.identifier().to_owned())
             .updates(existing_updates)
             .requirements(existing_requirements)
-            .build();
-
-        catalog.update_table(table_commit).await
+            .build())
     }
 }
 
@@ -260,7 +269,7 @@ mod tests {
     use crate::table::Table;
     use crate::test_utils::{make_encrypted_table, test_runtime};
     use crate::transaction::{ApplyTransactionAction, Transaction};
-    use crate::{Catalog, Error, ErrorKind, TableCreation, TableIdent};
+    use crate::{Catalog, Error, ErrorKind, TableCreation, TableIdent, TableUpdate};
 
     pub fn make_v1_table() -> Table {
         let file = File::open(format!(
@@ -476,6 +485,22 @@ mod tests {
         assert!(result.is_ok(), "Transaction should eventually succeed");
     }
 
+    #[tokio::test]
+    async fn test_prepare_commit() {
+        let table = make_v2_table();
+        let mut tx = create_test_transaction(&table);
+
+        let mut commit = tx.prepare_commit().await.unwrap();
+
+        assert_eq!(commit.identifier(), table.identifier());
+        assert_eq!(commit.take_updates(), vec![
+            TableUpdate::SetProperties {
+                updates: HashMap::from([("test.key".to_string(), "test.value".to_string())]),
+            },
+            TableUpdate::RemoveProperties { removals: vec![] },
+        ]);
+    }
+
     #[tokio::test]
     async fn test_commit_non_retryable_error() {
         // Create a test table with retry properties