feat(catalog/rest): add client for atomic multi-table transaction commits

crates/catalog/rest/public-api.txt

@@ -29,6 +29,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>
@@ -185,6 +198,7 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::RenameTable
 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<()>
 pub async fn iceberg_catalog_rest::RestCatalog::invalidate_token(&self) -> iceberg::error::Result<()>
 pub async fn iceberg_catalog_rest::RestCatalog::regenerate_token(&self) -> iceberg::error::Result<()>
 impl core::fmt::Debug for iceberg_catalog_rest::RestCatalog

crates/catalog/rest/src/catalog.rs

@@ -41,9 +41,9 @@ use crate::client::{
     HttpClient, deserialize_catalog_response, deserialize_unexpected_catalog_error,
 };
 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
@@ -202,6 +202,10 @@ impl RestCatalogConfig {
         self.url_prefixed(&["tables", "rename"])
     }
 
+    fn transactions_commit_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"])
     }
@@ -503,6 +507,89 @@ impl RestCatalog {
     pub async fn regenerate_token(&self) -> Result<()> {
         self.context().await?.client.regenerate_token().await
     }
+
+    /// Commit updates to multiple tables in one atomic operation, via the
+    /// REST spec's `POST /v1/{prefix}/transactions/commit`.
+    ///
+    /// Build each table's changes with
+    /// [`Transaction::prepare_commit`](iceberg::transaction::Transaction::prepare_commit)
+    /// and submit them together; the catalog applies all of the changes or
+    /// none of them.
+    ///
+    /// Unlike [`Catalog::update_table`], this method does not retry: on a
+    /// conflict the base metadata of every prepared table may be stale, so
+    /// re-preparing the whole transaction is the caller's decision.
+    ///
+    /// Two contract points callers must handle:
+    ///
+    /// * **A success returns no refreshed metadata** (the endpoint responds
+    ///   200/204 with an empty body), so post-commit state is not applied to
+    ///   any in-memory `Table`. Reload each affected table from the catalog
+    ///   before building further transactions on top of it.
+    /// * **Error semantics are the atomicity surface.** HTTP 409 means the
+    ///   commit did not apply and is safe to retry after re-preparing (the
+    ///   error is marked retryable). HTTP 500/502/503/504 mean the commit
+    ///   state is *unknown* — it may or may not have applied — so these are
+    ///   NOT marked retryable: blindly resubmitting can double-apply. Reload
+    ///   the tables and inspect before retrying.
+    pub async fn commit_transaction(&self, commits: Vec<TableCommit>) -> Result<()> {
+        if commits.is_empty() {
+            return Ok(());
+        }
+
+        let context = self.context().await?;
+
+        let table_changes: Vec<CommitTableRequest> = commits
+            .into_iter()
+            .map(|mut commit| CommitTableRequest {
+                identifier: Some(commit.identifier().clone()),
+                requirements: commit.take_requirements(),
+                updates: commit.take_updates(),
+            })
+            .collect();
+
+        let request = context
+            .client
+            .request(Method::POST, context.config.transactions_commit_endpoint())
+            .json(&CommitTransactionRequest { table_changes })
+            .build()?;
+
+        let http_response = context.client.query_catalog(request).await?;
+
+        match http_response.status() {
+            StatusCode::OK | StatusCode::NO_CONTENT => Ok(()),
+            StatusCode::NOT_FOUND => Err(Error::new(
+                ErrorKind::TableNotFound,
+                "Tried to commit a transaction against 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 => Err(Error::new(
+                ErrorKind::Unexpected,
+                "An unknown server-side problem occurred; the commit state is unknown.",
+            )),
+            StatusCode::BAD_GATEWAY => Err(Error::new(
+                ErrorKind::Unexpected,
+                "A gateway or proxy received an invalid response from the upstream server; the commit state is unknown.",
+            )),
+            StatusCode::GATEWAY_TIMEOUT => Err(Error::new(
+                ErrorKind::Unexpected,
+                "A server-side gateway timeout occurred; the commit state is unknown.",
+            )),
+            StatusCode::SERVICE_UNAVAILABLE => Err(Error::new(
+                ErrorKind::Unexpected,
+                "The service is unavailable; the commit state is unknown.",
+            )),
+            _ => Err(deserialize_unexpected_catalog_error(
+                http_response,
+                context.client.disable_header_redaction(),
+            )
+            .await),
+        }
+    }
 }
 
 /// All requests and expected responses are derived from the REST catalog API spec:
@@ -2958,4 +3045,105 @@ mod tests {
             assert_eq!(err.message(), "Catalog uri is required");
         }
     }
+
+    fn table_for_transaction(ident: &str) -> Table {
+        let file = File::open(format!(
+            "{}/testdata/{}",
+            env!("CARGO_MANIFEST_DIR"),
+            "create_table_response.json"
+        ))
+        .unwrap();
+        let reader = BufReader::new(file);
+        let resp = serde_json::from_reader::<_, LoadTableResult>(reader).unwrap();
+
+        Table::builder()
+            .metadata(resp.metadata)
+            .metadata_location(resp.metadata_location.unwrap())
+            .identifier(TableIdent::from_strs(["ns1", ident]).unwrap())
+            .file_io(FileIO::new_with_fs())
+            .runtime(test_runtime())
+            .build()
+            .unwrap()
+    }
+
+    async fn prepared_commits_for_two_tables() -> Vec<iceberg::TableCommit> {
+        let mut commits = Vec::new();
+        for ident in ["test1", "test2"] {
+            let table = table_for_transaction(ident);
+            let tx = Transaction::new(&table);
+            let tx = tx
+                .update_table_properties()
+                .set("owner".to_string(), ident.to_string())
+                .apply(tx)
+                .unwrap();
+            commits.push(tx.prepare_commit().await.unwrap());
+        }
+        commits
+    }
+
+    #[tokio::test]
+    async fn test_commit_transaction() {
+        let mut server = Server::new_async().await;
+        let config_mock = create_config_mock(&mut server).await;
+
+        let commit_transaction_mock = server
+            .mock("POST", "/v1/transactions/commit")
+            .match_body(mockito::Matcher::AllOf(vec![
+                mockito::Matcher::Regex("table-changes".to_string()),
+                mockito::Matcher::Regex("test1".to_string()),
+                mockito::Matcher::Regex("test2".to_string()),
+            ]))
+            .with_status(204)
+            .create_async()
+            .await;
+
+        let catalog = RestCatalog::new(
+            RestCatalogConfig::builder().uri(server.url()).build(),
+            Some(Arc::new(LocalFsStorageFactory)),
+            Runtime::current(),
+        );
+
+        let commits = prepared_commits_for_two_tables().await;
+        catalog.commit_transaction(commits).await.unwrap();
+
+        config_mock.assert_async().await;
+        commit_transaction_mock.assert_async().await;
+    }
+
+    #[tokio::test]
+    async fn test_commit_transaction_conflict_is_retryable() {
+        let mut server = Server::new_async().await;
+        let _config_mock = create_config_mock(&mut server).await;
+
+        let _commit_transaction_mock = server
+            .mock("POST", "/v1/transactions/commit")
+            .with_status(409)
+            .create_async()
+            .await;
+
+        let catalog = RestCatalog::new(
+            RestCatalogConfig::builder().uri(server.url()).build(),
+            Some(Arc::new(LocalFsStorageFactory)),
+            Runtime::current(),
+        );
+
+        let commits = prepared_commits_for_two_tables().await;
+        let err = catalog.commit_transaction(commits).await.unwrap_err();
+        assert_eq!(err.kind(), ErrorKind::CatalogCommitConflicts);
+        assert!(err.retryable());
+    }
+
+    #[tokio::test]
+    async fn test_commit_transaction_empty_is_noop() {
+        let server = Server::new_async().await;
+
+        let catalog = RestCatalog::new(
+            RestCatalogConfig::builder().uri(server.url()).build(),
+            Some(Arc::new(LocalFsStorageFactory)),
+            Runtime::current(),
+        );
+
+        // No HTTP mocks: an empty commit list must not contact the server.
+        catalog.commit_transaction(vec![]).await.unwrap();
+    }
 }

crates/catalog/rest/src/types.rs

@@ -289,6 +289,18 @@ 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 in an atomic operation.
+///
+/// Sent to `POST /v1/{prefix}/transactions/commit`. Each entry carries one
+/// table's requirements and updates; the catalog must apply all of the
+/// changes or none of them. Every entry's `identifier` must be present.
+pub struct CommitTransactionRequest {
+    /// Per-table changes to apply atomically.
+    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

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

crates/iceberg/src/transaction/mod.rs

@@ -203,6 +203,50 @@ impl Transaction {
         .1
     }
 
+    /// Run this transaction's actions against its base table and return the
+    /// resulting [`TableCommit`] without sending it to a catalog.
+    ///
+    /// This is the per-table building block for atomic multi-table commits:
+    /// prepare one `TableCommit` per table, then submit the whole group in a
+    /// single catalog operation (for example the REST catalog's
+    /// `/v1/{prefix}/transactions/commit` endpoint), which applies all of the
+    /// changes or none of them.
+    ///
+    /// Unlike [`Transaction::commit`], this method does not refresh the base
+    /// table and does not retry: requirements in the returned commit are
+    /// computed against the table state this transaction was created from,
+    /// and conflict handling across the transaction group belongs to the
+    /// caller.
+    pub async fn prepare_commit(self) -> Result<TableCommit> {
+        if self.actions.is_empty() {
+            return Err(Error::new(
+                ErrorKind::PreconditionFailed,
+                "No actions to prepare for commit",
+            ));
+        }
+
+        let mut current_table = self.table.clone();
+        let mut existing_updates: Vec<TableUpdate> = vec![];
+        let mut existing_requirements: Vec<TableRequirement> = vec![];
+
+        for action in &self.actions {
+            let action_commit = Arc::clone(action).commit(&current_table).await?;
+            // apply action commit to current_table
+            current_table = Self::apply(
+                current_table,
+                action_commit,
+                &mut existing_updates,
+                &mut existing_requirements,
+            )?;
+        }
+
+        Ok(TableCommit::builder()
+            .ident(self.table.identifier().to_owned())
+            .updates(existing_updates)
+            .requirements(existing_requirements)
+            .build())
+    }
+
     fn build_backoff(props: TableProperties) -> Result<ExponentialBackoff> {
         Ok(ExponentialBuilder::new()
             .with_min_delay(Duration::from_millis(props.commit_min_retry_wait_ms))
@@ -631,6 +675,30 @@ mod tests {
             err.message()
         );
     }
+
+    #[tokio::test]
+    async fn test_prepare_commit_returns_table_commit_without_catalog() {
+        let table = make_v2_minimal_table();
+        let tx = create_test_transaction(&table);
+
+        let mut table_commit = tx.prepare_commit().await.unwrap();
+
+        assert_eq!(table_commit.identifier(), table.identifier());
+        let updates = table_commit.take_updates();
+        assert!(
+            updates
+                .iter()
+                .any(|u| matches!(u, crate::TableUpdate::SetProperties { updates } if updates.contains_key("test.key"))),
+            "prepared commit should carry the SetProperties update"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_prepare_commit_empty_transaction_errors() {
+        let table = make_v2_minimal_table();
+        let tx = Transaction::new(&table);
+        assert!(tx.prepare_commit().await.is_err());
+    }
 }
 
 #[cfg(test)]