SS-148 make Transaction::commit internals public so Mz can bypass the built-in retry

crates/iceberg/public-api.txt

@@ -3076,14 +3076,26 @@ pub fn iceberg::transaction::AddColumn::optional(name: impl alloc::string::ToStr
 pub fn iceberg::transaction::AddColumn::required(name: impl alloc::string::ToString, field_type: iceberg::spec::Type, initial_default: iceberg::spec::Literal) -> Self
 impl iceberg::transaction::AddColumn
 pub fn iceberg::transaction::AddColumn::builder() -> AddColumnBuilder<((), (), (), (), (), (), ())>
+pub struct iceberg::transaction::RowDeltaAction
+impl iceberg::transaction::RowDeltaAction
+pub fn iceberg::transaction::RowDeltaAction::add_data_files(self, data_files: impl core::iter::traits::collect::IntoIterator<Item = iceberg::spec::DataFile>) -> Self
+pub fn iceberg::transaction::RowDeltaAction::add_delete_files(self, delete_files: impl core::iter::traits::collect::IntoIterator<Item = iceberg::spec::DataFile>) -> Self
+pub fn iceberg::transaction::RowDeltaAction::new() -> Self
+pub fn iceberg::transaction::RowDeltaAction::set_commit_uuid(self, commit_uuid: uuid::Uuid) -> Self
+pub fn iceberg::transaction::RowDeltaAction::set_snapshot_properties(self, snapshot_properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> Self
+pub fn iceberg::transaction::RowDeltaAction::with_check_duplicate(self, v: bool) -> Self
+impl core::default::Default for iceberg::transaction::RowDeltaAction
+pub fn iceberg::transaction::RowDeltaAction::default() -> Self
+impl iceberg::transaction::TransactionAction for iceberg::transaction::RowDeltaAction
+pub fn iceberg::transaction::RowDeltaAction::commit<'life0, 'async_trait>(self: alloc::sync::Arc<Self>, table: &'life0 iceberg::table::Table) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::transaction::ActionCommit>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub struct iceberg::transaction::Transaction
 impl iceberg::transaction::Transaction
 pub async fn iceberg::transaction::Transaction::commit(self, catalog: &dyn iceberg::Catalog) -> iceberg::Result<iceberg::table::Table>
 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 fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::sort_order::ReplaceSortOrderAction
-pub fn iceberg::transaction::Transaction::row_delta(&self) -> iceberg::transaction::row_delta::RowDeltaAction
+pub fn iceberg::transaction::Transaction::row_delta(&self) -> iceberg::transaction::RowDeltaAction
 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
 pub fn iceberg::transaction::Transaction::update_statistics(&self) -> iceberg::transaction::update_statistics::UpdateStatisticsAction
@@ -3093,8 +3105,12 @@ impl core::clone::Clone for iceberg::transaction::Transaction
 pub fn iceberg::transaction::Transaction::clone(&self) -> iceberg::transaction::Transaction
 pub trait iceberg::transaction::ApplyTransactionAction
 pub fn iceberg::transaction::ApplyTransactionAction::apply(self, tx: iceberg::transaction::Transaction) -> iceberg::Result<iceberg::transaction::Transaction>
-impl<T: TransactionAction + 'static> iceberg::transaction::ApplyTransactionAction for T
-pub fn T::apply(self, tx: iceberg::transaction::Transaction) -> iceberg::Result<iceberg::transaction::Transaction> where Self: core::marker::Sized
+impl<T: iceberg::transaction::TransactionAction + 'static> iceberg::transaction::ApplyTransactionAction for T
+pub fn T::apply(self, tx: iceberg::transaction::Transaction) -> core::result::Result<iceberg::transaction::Transaction, iceberg::Error>
+pub trait iceberg::transaction::TransactionAction: as_any::AsAny + core::marker::Sync + core::marker::Send
+pub fn iceberg::transaction::TransactionAction::commit<'life0, 'async_trait>(self: alloc::sync::Arc<Self>, table: &'life0 iceberg::table::Table) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::transaction::ActionCommit>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
+impl iceberg::transaction::TransactionAction for iceberg::transaction::RowDeltaAction
+pub fn iceberg::transaction::RowDeltaAction::commit<'life0, 'async_trait>(self: alloc::sync::Arc<Self>, table: &'life0 iceberg::table::Table) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::transaction::ActionCommit>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub mod iceberg::transform
 pub trait iceberg::transform::TransformFunction: core::marker::Send + core::marker::Sync + core::fmt::Debug
 pub fn iceberg::transform::TransformFunction::transform(&self, input: arrow_array::array::ArrayRef) -> iceberg::Result<arrow_array::array::ArrayRef>

crates/iceberg/src/catalog/mod.rs

@@ -343,11 +343,11 @@ pub struct TableCreation {
 
 /// TableCommit represents the commit of a table in the catalog.
 ///
-/// The builder is marked as private since it's dangerous and error-prone to construct
-/// [`TableCommit`] directly.
-/// Users are supposed to use [`crate::transaction::Transaction`] to update table.
+/// Most users should use [`crate::transaction::Transaction`] to update a table,
+/// which builds this internally. Construct it directly only when managing the
+/// commit yourself, from the updates and requirements produced by a
+/// [`crate::transaction::TransactionAction`].
 #[derive(Debug, TypedBuilder)]
-#[builder(build_method(vis = "pub(crate)"))]
 pub struct TableCommit {
     /// The table ident.
     ident: TableIdent,

crates/iceberg/src/transaction/action.rs

@@ -34,9 +34,16 @@ pub(crate) type BoxedTransactionAction = Arc<dyn TransactionAction>;
 /// Each action is responsible for generating the updates and requirements needed
 /// to modify the table metadata.
 #[async_trait]
-pub(crate) trait TransactionAction: AsAny + Sync + Send {
+pub trait TransactionAction: AsAny + Sync + Send {
     /// Commits this action against the provided table and returns the resulting updates.
-    /// NOTE: This function is intended for internal use only and should not be called directly by users.
+    /// NOTE: Most users should apply actions through [`Transaction`], which handles
+    /// rebasing onto the latest table state and retrying on commit conflicts. Call
+    /// this directly only to take over that responsibility yourself, passing the
+    /// resulting updates and requirements to [`Catalog::update_table`] via
+    /// [`TableCommit`].
+    ///
+    /// [`Catalog::update_table`]: crate::Catalog::update_table
+    /// [`TableCommit`]: crate::TableCommit
     ///
     /// # Arguments
     ///

crates/iceberg/src/transaction/mod.rs

@@ -56,6 +56,7 @@ pub use action::*;
 mod append;
 mod expire_snapshots;
 mod row_delta;
+pub use row_delta::RowDeltaAction;
 mod snapshot;
 mod sort_order;
 mod update_location;
@@ -76,7 +77,6 @@ use crate::table::Table;
 use crate::transaction::action::BoxedTransactionAction;
 use crate::transaction::append::FastAppendAction;
 use crate::transaction::expire_snapshots::ExpireSnapshotsAction;
-use crate::transaction::row_delta::RowDeltaAction;
 use crate::transaction::sort_order::ReplaceSortOrderAction;
 use crate::transaction::update_location::UpdateLocationAction;
 use crate::transaction::update_properties::UpdatePropertiesAction;

crates/iceberg/src/transaction/row_delta.rs

@@ -61,8 +61,15 @@ pub struct RowDeltaAction {
     added_delete_files: Vec<DataFile>,
 }
 
+impl Default for RowDeltaAction {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl RowDeltaAction {
-    pub(crate) fn new() -> Self {
+    /// Create a new row delta action. Equivalent to [`crate::transaction::Transaction::row_delta`].
+    pub fn new() -> Self {
         Self {
             check_duplicate: true,
             commit_uuid: None,