diff --git a/crates/catalog/rest/public-api.txt b/crates/catalog/rest/public-api.txt index 027df29b24..bad7f2188d 100644 --- a/crates/catalog/rest/public-api.txt +++ b/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::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_catalog_rest::CommitTransactionRequest +pub iceberg_catalog_rest::CommitTransactionRequest::table_changes: alloc::vec::Vec +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::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 @@ -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::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::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 diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index 0626ce5061..43b8146b8f 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/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) -> Result<()> { + if commits.is_empty() { + return Ok(()); + } + + let context = self.context().await?; + + let table_changes: Vec = 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 { + 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(); + } } diff --git a/crates/catalog/rest/src/types.rs b/crates/catalog/rest/src/types.rs index 21749fe0a4..c62828b365 100644 --- a/crates/catalog/rest/src/types.rs +++ b/crates/catalog/rest/src/types.rs @@ -289,6 +289,18 @@ pub struct CommitTableRequest { pub updates: Vec, } +#[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, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "kebab-case")] /// Response returned when a table is successfully updated. diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 41d4d7691b..e9a416e773 100644 --- a/crates/iceberg/public-api.txt +++ b/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 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 diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index d78f41cd42..d2f95a4f8f 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/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 { + 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 = vec![]; + let mut existing_requirements: Vec = vec![]; + + for action in &self.actions { + let action_commit = Arc::clone(action).commit(¤t_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 { 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)]