This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3265-1270e329b517a88bdb8eb641b4a59f45bdc0e872 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 0a8e9d1c813983c497adc193122b74052326f1c0 Author: Shawn Chang <[email protected]> AuthorDate: Wed Oct 14 08:07:01 2026 +0000 feat(transaction): Introduce stateful TransactionAction trait (#3265) * feat(transaction): Introduce stateful TransactionAction trait Rework the TransactionAction trait per the stateful transaction RFC (milestone 1): each action declares an associated retry-persistent State type created via new_state(), commit() borrows the action immutably and its state mutably per attempt, and a consuming cleanup() plus CommitStatus define terminal-cleanup semantics (not yet invoked; wired in a later milestone). Transactions now store one entry per action, pairing immutable intent with exclusively-owned execution state behind an object-safe ErasedActionEntry adapter. Cloning a transaction creates a new execution of the action plan: actions are duplicated and paired with fresh state, so retry state is never shared between executions. All existing actions migrate with State = () and no-op cleanup. * chore: Regenerate public-api baseline for action Clone impls * better comments * address comments * address comments * Address comments, add stateless tx action for convenience * clean up * rebase * minor * minor * fix import * actually test states reset and carried * address comments * update comments * fix clippy * testing --- crates/iceberg/public-api.txt | 22 +- crates/iceberg/src/test_utils/mod.rs | 2 + .../src/test_utils/{mod.rs => transaction.rs} | 26 +- crates/iceberg/src/transaction/action.rs | 270 ++++++++++++++++++++- crates/iceberg/src/transaction/append.rs | 44 ++-- crates/iceberg/src/transaction/expire_snapshots.rs | 38 +-- crates/iceberg/src/transaction/mod.rs | 9 +- crates/iceberg/src/transaction/sort_order.rs | 39 +-- crates/iceberg/src/transaction/update_location.rs | 22 +- .../iceberg/src/transaction/update_properties.rs | 15 +- crates/iceberg/src/transaction/update_schema.rs | 51 ++-- .../iceberg/src/transaction/update_statistics.rs | 24 +- .../src/transaction/upgrade_format_version.rs | 16 +- 13 files changed, 429 insertions(+), 149 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 466614757..b6d555595 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -3299,6 +3299,8 @@ pub struct iceberg::transaction::AddColumn impl iceberg::transaction::AddColumn pub fn iceberg::transaction::AddColumn::optional(name: impl alloc::string::ToString, field_type: iceberg::spec::Type) -> Self pub fn iceberg::transaction::AddColumn::required(name: impl alloc::string::ToString, field_type: iceberg::spec::Type, initial_default: iceberg::spec::Literal) -> Self +impl core::clone::Clone for iceberg::transaction::AddColumn +pub fn iceberg::transaction::AddColumn::clone(&self) -> iceberg::transaction::AddColumn impl iceberg::transaction::AddColumn pub fn iceberg::transaction::AddColumn::builder() -> AddColumnBuilder<((), (), (), (), (), (), ())> pub struct iceberg::transaction::ExpireSnapshotsAction @@ -3306,18 +3308,24 @@ impl iceberg::transaction::ExpireSnapshotsAction pub fn iceberg::transaction::ExpireSnapshotsAction::expire_older_than_ms(self, older_than_ms: i64) -> Self pub fn iceberg::transaction::ExpireSnapshotsAction::expire_snapshot_ids(self, snapshot_ids: impl core::iter::traits::collect::IntoIterator<Item = i64>) -> Self pub fn iceberg::transaction::ExpireSnapshotsAction::retain_last(self, retain_last: usize) -> Self +impl core::clone::Clone for iceberg::transaction::ExpireSnapshotsAction +pub fn iceberg::transaction::ExpireSnapshotsAction::clone(&self) -> iceberg::transaction::ExpireSnapshotsAction pub struct iceberg::transaction::FastAppendAction impl iceberg::transaction::FastAppendAction pub fn iceberg::transaction::FastAppendAction::add_data_files(self, data_files: impl core::iter::traits::collect::IntoIterator<Item = iceberg::spec::DataFile>) -> Self pub fn iceberg::transaction::FastAppendAction::set_commit_uuid(self, commit_uuid: uuid::Uuid) -> Self pub fn iceberg::transaction::FastAppendAction::set_snapshot_properties(self, snapshot_properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> Self pub fn iceberg::transaction::FastAppendAction::with_check_duplicate(self, v: bool) -> Self +impl core::clone::Clone for iceberg::transaction::FastAppendAction +pub fn iceberg::transaction::FastAppendAction::clone(&self) -> iceberg::transaction::FastAppendAction pub struct iceberg::transaction::ReplaceSortOrderAction impl iceberg::transaction::ReplaceSortOrderAction pub fn iceberg::transaction::ReplaceSortOrderAction::asc(self, name: &str, null_order: iceberg::spec::NullOrder) -> Self pub fn iceberg::transaction::ReplaceSortOrderAction::asc_with_transform(self, name: &str, transform: iceberg::spec::Transform, null_order: iceberg::spec::NullOrder) -> Self pub fn iceberg::transaction::ReplaceSortOrderAction::desc(self, name: &str, null_order: iceberg::spec::NullOrder) -> Self pub fn iceberg::transaction::ReplaceSortOrderAction::desc_with_transform(self, name: &str, transform: iceberg::spec::Transform, null_order: iceberg::spec::NullOrder) -> Self +impl core::clone::Clone for iceberg::transaction::ReplaceSortOrderAction +pub fn iceberg::transaction::ReplaceSortOrderAction::clone(&self) -> iceberg::transaction::ReplaceSortOrderAction 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> @@ -3330,29 +3338,37 @@ pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::trans pub fn iceberg::transaction::Transaction::update_statistics(&self) -> iceberg::transaction::UpdateStatisticsAction pub fn iceberg::transaction::Transaction::update_table_properties(&self) -> iceberg::transaction::UpdatePropertiesAction pub fn iceberg::transaction::Transaction::upgrade_table_version(&self) -> iceberg::transaction::UpgradeFormatVersionAction -impl core::clone::Clone for iceberg::transaction::Transaction -pub fn iceberg::transaction::Transaction::clone(&self) -> iceberg::transaction::Transaction pub struct iceberg::transaction::UpdateLocationAction impl iceberg::transaction::UpdateLocationAction pub fn iceberg::transaction::UpdateLocationAction::set_location(self, location: alloc::string::String) -> Self +impl core::clone::Clone for iceberg::transaction::UpdateLocationAction +pub fn iceberg::transaction::UpdateLocationAction::clone(&self) -> iceberg::transaction::UpdateLocationAction pub struct iceberg::transaction::UpdatePropertiesAction impl iceberg::transaction::UpdatePropertiesAction pub fn iceberg::transaction::UpdatePropertiesAction::remove(self, key: alloc::string::String) -> Self pub fn iceberg::transaction::UpdatePropertiesAction::set(self, key: alloc::string::String, value: alloc::string::String) -> Self +impl core::clone::Clone for iceberg::transaction::UpdatePropertiesAction +pub fn iceberg::transaction::UpdatePropertiesAction::clone(&self) -> iceberg::transaction::UpdatePropertiesAction pub struct iceberg::transaction::UpdateSchemaAction impl iceberg::transaction::UpdateSchemaAction pub fn iceberg::transaction::UpdateSchemaAction::add_column(self, add_column: iceberg::transaction::AddColumn) -> Self pub fn iceberg::transaction::UpdateSchemaAction::delete_column(self, name: impl alloc::string::ToString) -> Self +impl core::clone::Clone for iceberg::transaction::UpdateSchemaAction +pub fn iceberg::transaction::UpdateSchemaAction::clone(&self) -> iceberg::transaction::UpdateSchemaAction pub struct iceberg::transaction::UpdateStatisticsAction impl iceberg::transaction::UpdateStatisticsAction pub fn iceberg::transaction::UpdateStatisticsAction::remove_statistics(self, snapshot_id: i64) -> Self pub fn iceberg::transaction::UpdateStatisticsAction::set_statistics(self, statistics_file: iceberg::spec::StatisticsFile) -> Self +impl core::clone::Clone for iceberg::transaction::UpdateStatisticsAction +pub fn iceberg::transaction::UpdateStatisticsAction::clone(&self) -> iceberg::transaction::UpdateStatisticsAction pub struct iceberg::transaction::UpgradeFormatVersionAction impl iceberg::transaction::UpgradeFormatVersionAction pub fn iceberg::transaction::UpgradeFormatVersionAction::set_format_version(self, format_version: iceberg::spec::FormatVersion) -> Self +impl core::clone::Clone for iceberg::transaction::UpgradeFormatVersionAction +pub fn iceberg::transaction::UpgradeFormatVersionAction::clone(&self) -> iceberg::transaction::UpgradeFormatVersionAction 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 +impl<T: TransactionAction> iceberg::transaction::ApplyTransactionAction for T pub fn T::apply(self, tx: iceberg::transaction::Transaction) -> core::result::Result<iceberg::transaction::Transaction, iceberg::Error> pub mod iceberg::transform pub trait iceberg::transform::TransformFunction: core::marker::Send + core::marker::Sync + core::fmt::Debug diff --git a/crates/iceberg/src/test_utils/mod.rs b/crates/iceberg/src/test_utils/mod.rs index 753a17e53..94310dd37 100644 --- a/crates/iceberg/src/test_utils/mod.rs +++ b/crates/iceberg/src/test_utils/mod.rs @@ -26,8 +26,10 @@ mod encryption; mod record_batch; mod runtime; pub(crate) mod scan; +mod transaction; pub(crate) use delete_vector::encode_dv_blob; pub(crate) use encryption::{make_encrypted_table, make_encryption_manager}; pub(crate) use record_batch::check_record_batches; pub(crate) use runtime::test_runtime; +pub(crate) use transaction::commit_with_fresh_state; diff --git a/crates/iceberg/src/test_utils/mod.rs b/crates/iceberg/src/test_utils/transaction.rs similarity index 59% copy from crates/iceberg/src/test_utils/mod.rs copy to crates/iceberg/src/test_utils/transaction.rs index 753a17e53..292848359 100644 --- a/crates/iceberg/src/test_utils/mod.rs +++ b/crates/iceberg/src/test_utils/transaction.rs @@ -15,19 +15,15 @@ // specific language governing permissions and limitations // under the License. -//! Test utilities for this crate's own tests. -//! -//! Compiled under `cfg(test)` only, so nothing here is public API. Fixtures -//! that other crates in the workspace need from their own tests live in the -//! `iceberg_test_utils` crate instead. +use crate::error::Result; +use crate::table::Table; +use crate::transaction::{ActionCommit, TransactionAction}; -mod delete_vector; -mod encryption; -mod record_batch; -mod runtime; -pub(crate) mod scan; - -pub(crate) use delete_vector::encode_dv_blob; -pub(crate) use encryption::{make_encrypted_table, make_encryption_manager}; -pub(crate) use record_batch::check_record_batches; -pub(crate) use runtime::test_runtime; +/// Runs a single commit attempt of an action with fresh state. +pub(crate) async fn commit_with_fresh_state<A: TransactionAction>( + action: A, + table: &Table, +) -> Result<ActionCommit> { + let mut state = action.new_state(); + action.commit(&mut state, table).await +} diff --git a/crates/iceberg/src/transaction/action.rs b/crates/iceberg/src/transaction/action.rs index aa0a05d0d..46868c3a2 100644 --- a/crates/iceberg/src/transaction/action.rs +++ b/crates/iceberg/src/transaction/action.rs @@ -16,7 +16,6 @@ // under the License. use std::mem::take; -use std::sync::Arc; use as_any::AsAny; use async_trait::async_trait; @@ -25,28 +24,174 @@ use crate::table::Table; use crate::transaction::Transaction; use crate::{Result, TableRequirement, TableUpdate}; -/// A boxed, thread-safe reference to a `TransactionAction`. -pub(crate) type BoxedTransactionAction = Arc<dyn TransactionAction>; +/// A boxed entry pairing a transaction action with its retry-persistent state. +pub(crate) type TransactionActionEntry = Box<dyn DynActionEntry>; /// A trait representing an atomic action that can be part of a transaction. /// /// Implementors of this trait define how a specific action is committed to a table. /// Each action is responsible for generating the updates and requirements needed /// to modify the table metadata. +/// +/// An action's intent is immutable once applied to a transaction. Retry-persistent +/// state lives in the associated [`TransactionAction::State`], which survives replay +/// attempts of one logical execution but is never shared between executions. #[async_trait] -pub(crate) trait TransactionAction: AsAny + Sync + Send { +pub(crate) trait TransactionAction: Clone + Send + Sync + 'static { + /// Retry-persistent state exclusively owned by one logical execution of this + /// action. Stateless actions implement [`StatelessAction`] instead, which + /// supplies `State = ()`. + type State: Send + Sync + 'static; + + /// Creates fresh state for one logical execution. + /// + /// This is infallible and table-independent; table-dependent initialization + /// happens during [`TransactionAction::commit`]. + fn new_state(&self) -> Self::State; + /// 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. /// + /// One replay attempt: the action (intent) is borrowed immutably, its execution + /// state mutably, and the table reflects the current transaction-local base. + /// /// # Arguments /// + /// * `state` - The retry-persistent state for this logical execution. /// * `table` - The current state of the table this action should apply to. /// /// # Returns /// /// An `ActionCommit` containing table updates and table requirements, /// or an error if the commit fails. - async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit>; + async fn commit(&self, state: &mut Self::State, table: &Table) -> Result<ActionCommit>; + + /// Best-effort cleanup after the transaction reaches a terminal result. + /// + /// Consumes the action and its state: ownership makes any further use + /// of this action after cleanup a compile error. + /// Cleanup must not change the already-determined transaction result. + /// + /// See [`CommitStatus`] for what each terminal status allows cleanup to do. + // TODO: invoke this from `Transaction::commit` once terminal status + // classification is wired up (stateful transaction RFC, milestone 3). + #[allow(dead_code)] + async fn cleanup(self, state: Self::State, table: &Table, status: CommitStatus); +} + +/// Convenience trait for actions that carry no per-attempt state. +/// +/// Implementing `StatelessAction` yields a [`TransactionAction`] impl with +/// `State = ()`, a no-op [`TransactionAction::new_state`], a no-op +/// [`TransactionAction::cleanup`], and a [`TransactionAction::commit`] that +/// forwards to [`StatelessAction::commit`]. Actions that need retry-persistent +/// state (e.g. a persistent snapshot producer) implement `TransactionAction` +/// directly instead. +#[async_trait] +pub(crate) trait StatelessAction: Clone + Send + Sync + 'static { + /// One commit attempt against the current transaction-local table. + /// + /// Same contract as [`TransactionAction::commit`], minus the state parameter. + async fn commit(&self, table: &Table) -> Result<ActionCommit>; +} + +#[async_trait] +impl<T: StatelessAction> TransactionAction for T { + type State = (); + + fn new_state(&self) {} + + async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> { + StatelessAction::commit(self, table).await + } + + async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {} +} + +/// Classification of a transaction's terminal result, consumed by +/// [`TransactionAction::cleanup`] to determine what is safe to delete. +/// +/// The transaction/catalog layer determines this classification; actions +/// consume it rather than independently interpreting catalog errors. +// TODO: produce this classification in `Transaction::commit` once terminal +// cleanup is wired up (stateful transaction RFC, milestone 3). +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CommitStatus { + /// The catalog confirmed that the commit was applied. Cleanup may remove + /// owned artifacts not retained by the committed result. + Committed, + /// The transaction definitively did not commit and will not retry. + /// Owned artifacts are deletable. + Failed, + /// The commit request was submitted, but its outcome could not be + /// resolved: the catalog may or may not have applied it. Cleanup must + /// delete nothing, and the transaction must not be retried. + /// + /// Example: every action executed and validated successfully, but the + /// connection failed while awaiting the catalog's response to + /// `update_table`. + Unknown, +} + +/// An entry in a transaction, pairing an action's immutable intent with the +/// retry-persistent state of one logical execution. +/// +/// The pairing is preserved by construction: the entry is created with fresh +/// state and owns both exclusively, so terminal cleanup can consume them together. +pub(crate) struct ActionEntry<A: TransactionAction> { + action: A, + state: A::State, +} + +impl<A: TransactionAction> ActionEntry<A> { + fn new(action: A) -> Self { + let state = action.new_state(); + Self { action, state } + } + + /// The action (intent) held by this entry. + #[allow(dead_code)] + pub(crate) fn action(&self) -> &A { + &self.action + } +} + +/// Object-safe adapter over [`ActionEntry`], allowing a transaction to store +/// heterogeneous entries while keeping each action paired with its own state. +#[async_trait] +pub(crate) trait DynActionEntry: AsAny + Send + Sync { + /// One replay attempt against the current transaction-local table. + async fn commit(&mut self, table: &Table) -> Result<ActionCommit>; + + /// Best-effort terminal cleanup, consuming the entry. + // TODO: invoke from `Transaction::commit` once terminal status + // classification is wired up (stateful transaction RFC, milestone 3). + #[allow(dead_code)] + async fn cleanup(self: Box<Self>, table: &Table, status: CommitStatus); + + /// Creates a new entry for a new logical execution of the same action: + /// the action (intent) is cloned, and it is paired with fresh state from + /// [`TransactionAction::new_state`]. Accrued retry state is never carried + /// over to the new entry. + #[allow(dead_code)] + fn clone_with_new_state(&self) -> TransactionActionEntry; +} + +#[async_trait] +impl<A: TransactionAction> DynActionEntry for ActionEntry<A> { + async fn commit(&mut self, table: &Table) -> Result<ActionCommit> { + self.action.commit(&mut self.state, table).await + } + + async fn cleanup(self: Box<Self>, table: &Table, status: CommitStatus) { + let ActionEntry { action, state } = *self; + action.cleanup(state, table, status).await + } + + fn clone_with_new_state(&self) -> TransactionActionEntry { + Box::new(ActionEntry::new(self.action.clone())) + } } /// A helper trait for applying a `TransactionAction` to a `Transaction`. @@ -66,10 +211,10 @@ pub trait ApplyTransactionAction { fn apply(self, tx: Transaction) -> Result<Transaction>; } -impl<T: TransactionAction + 'static> ApplyTransactionAction for T { +impl<T: TransactionAction> ApplyTransactionAction for T { fn apply(self, mut tx: Transaction) -> Result<Transaction> where Self: Sized { - tx.actions.push(Arc::new(self)); + tx.actions.push(Box::new(ActionEntry::new(self))); Ok(tx) } } @@ -106,23 +251,31 @@ impl ActionCommit { #[cfg(test)] mod tests { use std::str::FromStr; - use std::sync::Arc; use as_any::Downcast; use async_trait::async_trait; use uuid::Uuid; use crate::table::Table; + use crate::test_utils::commit_with_fresh_state; use crate::transaction::Transaction; - use crate::transaction::action::{ActionCommit, ApplyTransactionAction, TransactionAction}; + use crate::transaction::action::{ + ActionCommit, ActionEntry, ApplyTransactionAction, CommitStatus, DynActionEntry, + StatelessAction, TransactionAction, + }; use crate::transaction::tests::make_v2_table; use crate::{Result, TableRequirement, TableUpdate}; + #[derive(Clone)] struct TestAction; #[async_trait] impl TransactionAction for TestAction { - async fn commit(self: Arc<Self>, _table: &Table) -> Result<ActionCommit> { + type State = (); + + fn new_state(&self) -> Self::State {} + + async fn commit(&self, _state: &mut (), _table: &Table) -> Result<ActionCommit> { Ok(ActionCommit::new( vec![TableUpdate::SetLocation { location: String::from("s3://bucket/prefix/table/"), @@ -132,6 +285,42 @@ mod tests { }], )) } + + async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {} + } + + #[derive(Clone)] + struct CountingAction; + + #[async_trait] + impl TransactionAction for CountingAction { + type State = u32; + + fn new_state(&self) -> Self::State { + 0 + } + + async fn commit(&self, state: &mut u32, _table: &Table) -> Result<ActionCommit> { + *state += 1; + Ok(ActionCommit::new(vec![], vec![])) + } + + async fn cleanup(self, _state: u32, _table: &Table, _status: CommitStatus) {} + } + + #[derive(Clone)] + struct StatelessTestAction; + + #[async_trait] + impl StatelessAction for StatelessTestAction { + async fn commit(&self, _table: &Table) -> Result<ActionCommit> { + Ok(ActionCommit::new( + vec![TableUpdate::SetLocation { + location: String::from("s3://bucket/prefix/stateless/"), + }], + vec![], + )) + } } #[tokio::test] @@ -139,7 +328,7 @@ mod tests { let table = make_v2_table(); let action = TestAction; - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let requirements = action_commit.take_requirements(); @@ -159,14 +348,69 @@ mod tests { let tx = Transaction::new(&table); let updated_tx = action.apply(tx).unwrap(); - // There should be one action in the transaction now + // There should be one action entry in the transaction now assert_eq!(updated_tx.actions.len(), 1); (*updated_tx.actions[0]) - .downcast_ref::<TestAction>() + .downcast_ref::<ActionEntry<TestAction>>() .expect("TestAction was not applied to Transaction!"); } + #[tokio::test] + async fn test_entry_retains_state_across_commits() { + let table = make_v2_table(); + let mut tx = CountingAction.apply(Transaction::new(&table)).unwrap(); + + let entry = &mut tx.actions[0]; + entry.commit(&table).await.unwrap(); + entry.commit(&table).await.unwrap(); + + let entry = (**entry) + .downcast_ref::<ActionEntry<CountingAction>>() + .unwrap(); + assert_eq!(entry.state, 2); + } + + #[tokio::test] + async fn test_clone_with_new_state_resets_state() { + let table = make_v2_table(); + let mut tx = CountingAction.apply(Transaction::new(&table)).unwrap(); + tx.actions[0].commit(&table).await.unwrap(); + + let cloned = tx.actions[0].clone_with_new_state(); + + let original = (*tx.actions[0]) + .downcast_ref::<ActionEntry<CountingAction>>() + .unwrap(); + let cloned = (*cloned) + .downcast_ref::<ActionEntry<CountingAction>>() + .unwrap(); + assert_eq!(original.state, 1); + assert_eq!(cloned.state, 0); + } + + #[tokio::test] + async fn test_stateless_action_blanket_impl() { + let table = make_v2_table(); + + // commit goes through the blanket TransactionAction impl + let mut commit = commit_with_fresh_state(StatelessTestAction, &table) + .await + .unwrap(); + assert_eq!(commit.take_updates(), vec![TableUpdate::SetLocation { + location: String::from("s3://bucket/prefix/stateless/"), + }]); + + // ApplyTransactionAction is available through the blanket impl too + let tx = StatelessTestAction.apply(Transaction::new(&table)).unwrap(); + assert_eq!(tx.actions.len(), 1); + + // and the erased cleanup forwarding chain compiles and runs + let entry: Box<ActionEntry<StatelessTestAction>> = + Box::new(ActionEntry::new(StatelessTestAction)); + entry.cleanup(&table, CommitStatus::Committed).await; + } + #[test] fn test_action_commit() { // Create dummy updates and requirements diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 3260f9ffa..2b6f01044 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -16,7 +16,6 @@ // under the License. use std::collections::{HashMap, HashSet}; -use std::sync::Arc; use async_trait::async_trait; use uuid::Uuid; @@ -24,12 +23,14 @@ use uuid::Uuid; use crate::error::Result; use crate::spec::{DataFile, ManifestEntry, ManifestFile, Operation}; use crate::table::Table; +use crate::transaction::action::CommitStatus; use crate::transaction::snapshot::{ DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, }; use crate::transaction::{ActionCommit, TransactionAction}; /// FastAppendAction is a transaction action for fast append data files to the table. +#[derive(Clone)] pub struct FastAppendAction { check_duplicate: bool, // below are properties used to create SnapshotProducer when commit @@ -87,7 +88,13 @@ impl FastAppendAction { #[async_trait] impl TransactionAction for FastAppendAction { - async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + // TODO: replace with a persistent `SimpleSnapshotProducer` once snapshot + // production is migrated (stateful transaction RFC, milestone 2). + type State = (); + + fn new_state(&self) -> Self::State {} + + async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> { let snapshot_producer = SnapshotProducer::new( table, self.commit_uuid.unwrap_or_else(Uuid::now_v7), @@ -107,6 +114,11 @@ impl TransactionAction for FastAppendAction { .commit(FastAppendOperation, DefaultManifestProcess) .await } + + async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) { + // Stateless for now: generated manifests are not yet tracked for + // terminal cleanup (stateful transaction RFC, milestone 2/3). + } } struct FastAppendOperation; @@ -170,9 +182,9 @@ mod tests { Struct, TableMetadata, }; use crate::table::Table; - use crate::test_utils::{make_encrypted_table, test_runtime}; + use crate::test_utils::{commit_with_fresh_state, make_encrypted_table, test_runtime}; + use crate::transaction::Transaction; use crate::transaction::tests::make_v2_minimal_table; - use crate::transaction::{Transaction, TransactionAction}; use crate::{TableIdent, TableRequirement, TableUpdate}; fn render_template(template: &str, ctx: Value) -> String { @@ -341,7 +353,7 @@ mod tests { let tx = Transaction::new(&table); let action = tx.fast_append().add_data_files(vec![new_file]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { @@ -409,7 +421,7 @@ mod tests { let tx = Transaction::new(&table); let action = tx.fast_append().add_data_files(vec![new_file]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_snapshot: SnapshotRef = updates @@ -485,7 +497,7 @@ mod tests { let table = make_v2_minimal_table(); let tx = Transaction::new(&table); let action = tx.fast_append().add_data_files(vec![]); - assert!(Arc::new(action).commit(&table).await.is_err()); + assert!(commit_with_fresh_state(action, &table).await.is_err()); } /// A `fast_append` must write the manifest list and the manifest @@ -522,7 +534,7 @@ mod tests { let tx = Transaction::new(&table); let action = tx.fast_append().add_data_files(vec![data_file]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { @@ -582,7 +594,7 @@ mod tests { .fast_append() .set_snapshot_properties(snapshot_properties) .add_data_files(vec![data_file]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); // Check customized properties is contained in snapshot summary properties. @@ -666,7 +678,7 @@ mod tests { let tx = Transaction::new(&table); let action = tx.fast_append().add_data_files(vec![data_file]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let added_key_ids: Vec<String> = updates @@ -741,7 +753,7 @@ mod tests { .set_snapshot_properties(snapshot_properties) .add_data_files(vec![data_file]); // Must not panic during total computation. - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { @@ -775,7 +787,7 @@ mod tests { let action = tx .fast_append() .set_snapshot_properties(snapshot_properties); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); // Check customized properties is contained in snapshot summary properties. @@ -814,7 +826,7 @@ mod tests { let action = action.add_data_files(vec![data_file.clone()]); - assert!(Arc::new(action).commit(&table).await.is_err()); + assert!(commit_with_fresh_state(action, &table).await.is_err()); } #[tokio::test] @@ -841,7 +853,7 @@ mod tests { make_file(200, 20), make_file(300, 30), ]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let files = committed_data_files(&table, &action_commit.take_updates()).await; assert_eq!(1, files.len()); assert_eq!(100, files[0].file_size_in_bytes()); @@ -870,7 +882,7 @@ mod tests { .fast_append() .with_check_duplicate(false) .add_data_files(vec![make_file(), make_file()]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let files = committed_data_files(&table, &action_commit.take_updates()).await; assert_eq!(1, files.len()); } @@ -893,7 +905,7 @@ mod tests { .unwrap(); let action = action.add_data_files(vec![data_file.clone()]); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let requirements = action_commit.take_requirements(); diff --git a/crates/iceberg/src/transaction/expire_snapshots.rs b/crates/iceberg/src/transaction/expire_snapshots.rs index 0e08c6b24..13813fd7f 100644 --- a/crates/iceberg/src/transaction/expire_snapshots.rs +++ b/crates/iceberg/src/transaction/expire_snapshots.rs @@ -16,7 +16,6 @@ // under the License. use std::collections::HashSet; -use std::sync::Arc; use async_trait::async_trait; use chrono::Utc; @@ -26,7 +25,7 @@ use crate::spec::{ MAIN_BRANCH, SnapshotReference, SnapshotRetention, TableMetadata, TableProperties, }; use crate::table::Table; -use crate::transaction::action::{ActionCommit, TransactionAction}; +use crate::transaction::action::{ActionCommit, StatelessAction}; use crate::{Error, Result, TableRequirement, TableUpdate}; /// A transaction action that removes snapshots from table metadata. @@ -52,6 +51,7 @@ use crate::{Error, Result, TableRequirement, TableUpdate}; /// explicitly is an error, since /// [`remove_snapshots`](crate::spec::TableMetadataBuilder::remove_snapshots) would otherwise /// drop the ref silently. +#[derive(Clone)] pub struct ExpireSnapshotsAction { explicit_ids_to_remove: Vec<i64>, older_than_ms: Option<i64>, @@ -296,8 +296,8 @@ struct ExpirePlan { } #[async_trait] -impl TransactionAction for ExpireSnapshotsAction { - async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { +impl StatelessAction for ExpireSnapshotsAction { + async fn commit(&self, table: &Table) -> Result<ActionCommit> { let metadata = table.metadata(); let properties = metadata.table_properties(); @@ -372,8 +372,9 @@ mod tests { SnapshotRetention, StatisticsFile, Summary, }; use crate::table::Table; + use crate::test_utils::commit_with_fresh_state; use crate::transaction::Transaction; - use crate::transaction::action::{ApplyTransactionAction, TransactionAction}; + use crate::transaction::action::ApplyTransactionAction; use crate::transaction::expire_snapshots::ExpireSnapshotsAction; use crate::transaction::tests::{make_v2_minimal_table, make_v2_table}; use crate::{TableRequirement, TableUpdate}; @@ -397,7 +398,10 @@ mod tests { } async fn updates_of(table: &Table, action: ExpireSnapshotsAction) -> Vec<TableUpdate> { - Arc::new(action).commit(table).await.unwrap().take_updates() + commit_with_fresh_state(action, table) + .await + .unwrap() + .take_updates() } async fn expired(table: &Table, action: ExpireSnapshotsAction) -> Vec<i64> { @@ -580,7 +584,7 @@ mod tests { async fn test_cannot_expire_current_snapshot() { let table = make_v2_table(); let action = action().expire_snapshot_ids(vec![CURRENT_SNAPSHOT]); - assert!(Arc::new(action).commit(&table).await.is_err()); + assert!(commit_with_fresh_state(action, &table).await.is_err()); } /// `make_v2_table` with a tag pointing at the older snapshot. @@ -607,14 +611,13 @@ mod tests { async fn test_cannot_expire_tagged_snapshot_explicitly() { let table = table_with_tag_on_old(); let action = action().expire_snapshot_ids(vec![OLD_SNAPSHOT]); - assert!(Arc::new(action).commit(&table).await.is_err()); + assert!(commit_with_fresh_state(action, &table).await.is_err()); } #[tokio::test] async fn test_age_expiry_skips_tagged_snapshot() { let table = table_with_tag_on_old(); - let mut commit = Arc::new(action().expire_older_than_ms(i64::MAX)) - .commit(&table) + let mut commit = commit_with_fresh_state(action().expire_older_than_ms(i64::MAX), &table) .await .unwrap(); // Both snapshots are referenced (current + tag), so nothing is expired. @@ -765,16 +768,16 @@ mod tests { let table = table.with_metadata(Arc::new(metadata)); let action = action().expire_snapshot_ids(vec![OLD_SNAPSHOT]); - assert!(Arc::new(action).commit(&table).await.is_err()); + assert!(commit_with_fresh_state(action, &table).await.is_err()); } #[tokio::test] async fn test_commit_asserts_main_ref() { let table = make_v2_table(); - let mut commit = Arc::new(action().expire_snapshot_ids(vec![OLD_SNAPSHOT])) - .commit(&table) - .await - .unwrap(); + let mut commit = + commit_with_fresh_state(action().expire_snapshot_ids(vec![OLD_SNAPSHOT]), &table) + .await + .unwrap(); assert!( commit .take_requirements() @@ -902,8 +905,7 @@ mod tests { async fn test_retain_last_zero_errors() { let table = make_v2_table(); assert!( - Arc::new(action().retain_last(0)) - .commit(&table) + commit_with_fresh_state(action().retain_last(0), &table) .await .is_err() ); @@ -917,7 +919,7 @@ mod tests { ); // 2 is the head of a non-main branch, so it cannot be expired explicitly. let action = action().expire_snapshot_ids(vec![2]); - assert!(Arc::new(action).commit(&table).await.is_err()); + assert!(commit_with_fresh_state(action, &table).await.is_err()); } #[tokio::test] diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index 362247127..efae9d01b 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -72,7 +72,7 @@ pub use update_schema::AddColumn; use crate::error::Result; use crate::spec::TableProperties; use crate::table::Table; -use crate::transaction::action::BoxedTransactionAction; +use crate::transaction::action::TransactionActionEntry; pub use crate::transaction::append::FastAppendAction; pub use crate::transaction::expire_snapshots::ExpireSnapshotsAction; pub use crate::transaction::sort_order::ReplaceSortOrderAction; @@ -84,10 +84,9 @@ pub use crate::transaction::upgrade_format_version::UpgradeFormatVersionAction; use crate::{Catalog, TableCommit, TableRequirement, TableUpdate}; /// Table transaction. -#[derive(Clone)] pub struct Transaction { table: Table, - actions: Vec<BoxedTransactionAction>, + actions: Vec<TransactionActionEntry>, } impl Transaction { @@ -221,8 +220,8 @@ impl Transaction { 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(¤t_table).await?; + for entry in self.actions.iter_mut() { + let action_commit = entry.commit(¤t_table).await?; // apply action commit to current_table current_table = Self::apply( current_table, diff --git a/crates/iceberg/src/transaction/sort_order.rs b/crates/iceberg/src/transaction/sort_order.rs index fb4af4927..4543a1e2c 100644 --- a/crates/iceberg/src/transaction/sort_order.rs +++ b/crates/iceberg/src/transaction/sort_order.rs @@ -15,14 +15,12 @@ // specific language governing permissions and limitations // under the License. -use std::sync::Arc; - use async_trait::async_trait; use crate::error::{Result, invalid_data}; use crate::spec::{NullOrder, SchemaRef, SortDirection, SortField, SortOrder, Transform}; use crate::table::Table; -use crate::transaction::{ActionCommit, TransactionAction}; +use crate::transaction::{ActionCommit, StatelessAction}; use crate::{TableRequirement, TableUpdate}; /// Represents a sort field whose construction and validation are deferred until commit time. @@ -58,6 +56,7 @@ impl PendingSortField { } /// Transaction action for replacing sort order. +#[derive(Clone)] pub struct ReplaceSortOrderAction { pending_sort_fields: Vec<PendingSortField>, } @@ -137,8 +136,8 @@ impl ReplaceSortOrderAction { } #[async_trait] -impl TransactionAction for ReplaceSortOrderAction { - async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { +impl StatelessAction for ReplaceSortOrderAction { + async fn commit(&self, table: &Table) -> Result<ActionCommit> { let current_schema = table.metadata().current_schema(); let sort_fields: Result<Vec<SortField>> = self .pending_sort_fields @@ -172,16 +171,16 @@ impl TransactionAction for ReplaceSortOrderAction { #[cfg(test)] mod tests { - use std::sync::Arc; - use as_any::Downcast; use crate::catalog::Catalog; use crate::memory::tests::new_memory_catalog; use crate::spec::{NullOrder, SortDirection, Transform}; + use crate::test_utils::commit_with_fresh_state; + use crate::transaction::action::ActionEntry; use crate::transaction::sort_order::{PendingSortField, ReplaceSortOrderAction}; use crate::transaction::tests::{make_v2_table, make_v3_minimal_table_in_catalog}; - use crate::transaction::{ApplyTransactionAction, Transaction, TransactionAction}; + use crate::transaction::{ApplyTransactionAction, Transaction}; use crate::{ErrorKind, TableUpdate}; #[test] @@ -197,8 +196,9 @@ mod tests { .unwrap(); let replace_sort_order = (*tx.actions[0]) - .downcast_ref::<ReplaceSortOrderAction>() - .unwrap(); + .downcast_ref::<ActionEntry<ReplaceSortOrderAction>>() + .unwrap() + .action(); assert_eq!(replace_sort_order.pending_sort_fields, vec![ PendingSortField { @@ -229,8 +229,9 @@ mod tests { .unwrap(); let replace_sort_order = (*tx.actions[0]) - .downcast_ref::<ReplaceSortOrderAction>() - .unwrap(); + .downcast_ref::<ActionEntry<ReplaceSortOrderAction>>() + .unwrap() + .action(); assert_eq!(replace_sort_order.pending_sort_fields, vec![ PendingSortField { @@ -251,13 +252,13 @@ mod tests { #[tokio::test] async fn test_replace_sort_order_with_transform_commits() { let table = make_v2_table(); - let action = Arc::new(ReplaceSortOrderAction::new().asc_with_transform( + let action = ReplaceSortOrderAction::new().asc_with_transform( "x", Transform::Bucket(16), NullOrder::First, - )); + ); - let mut action_commit = TransactionAction::commit(action, &table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); assert_eq!(updates.len(), 2); @@ -272,13 +273,13 @@ mod tests { async fn test_replace_sort_order_rejects_incompatible_transform() { let table = make_v2_table(); // `x` is a `long` column; `year` only accepts date/timestamp types. - let action = Arc::new(ReplaceSortOrderAction::new().asc_with_transform( + let action = ReplaceSortOrderAction::new().asc_with_transform( "x", Transform::Year, NullOrder::First, - )); + ); - let err = TransactionAction::commit(action, &table) + let err = commit_with_fresh_state(action, &table) .await .err() .expect("year transform on a long column should be rejected"); @@ -293,7 +294,7 @@ mod tests { ReplaceSortOrderAction::new().asc_with_transform("x", transform, NullOrder::First), ReplaceSortOrderAction::new().desc_with_transform("x", transform, NullOrder::Last), ] { - let err = TransactionAction::commit(Arc::new(action), &table) + let err = commit_with_fresh_state(action, &table) .await .err() .expect("unknown and void sort transforms should be rejected"); diff --git a/crates/iceberg/src/transaction/update_location.rs b/crates/iceberg/src/transaction/update_location.rs index 7d0e2f954..147bd6062 100644 --- a/crates/iceberg/src/transaction/update_location.rs +++ b/crates/iceberg/src/transaction/update_location.rs @@ -15,13 +15,11 @@ // specific language governing permissions and limitations // under the License. -use std::sync::Arc; - use async_trait::async_trait; use crate::error::invalid_data; use crate::table::Table; -use crate::transaction::action::{ActionCommit, TransactionAction}; +use crate::transaction::action::{ActionCommit, StatelessAction}; use crate::{Result, TableUpdate}; /// A transaction action that sets or updates the location of a table. @@ -29,6 +27,7 @@ use crate::{Result, TableUpdate}; /// This action is used to explicitly set a new metadata location during a transaction, /// typically as part of advanced commit or recovery flows. The location is optional until /// explicitly set via [`UpdateLocationAction::set_location`]. +#[derive(Clone)] pub struct UpdateLocationAction { location: Option<String>, } @@ -55,11 +54,13 @@ impl UpdateLocationAction { } #[async_trait] -impl TransactionAction for UpdateLocationAction { - async fn commit(self: Arc<Self>, _table: &Table) -> Result<ActionCommit> { +impl StatelessAction for UpdateLocationAction { + async fn commit(&self, _table: &Table) -> Result<ActionCommit> { let updates: Vec<TableUpdate>; - if let Some(location) = self.location.clone() { - updates = vec![TableUpdate::SetLocation { location }]; + if let Some(location) = self.location.as_deref() { + updates = vec![TableUpdate::SetLocation { + location: location.to_owned(), + }]; } else { return Err(invalid_data!( "Location is not set for UpdateLocationAction!" @@ -75,7 +76,7 @@ mod tests { use as_any::Downcast; use crate::transaction::Transaction; - use crate::transaction::action::ApplyTransactionAction; + use crate::transaction::action::{ActionEntry, ApplyTransactionAction}; use crate::transaction::tests::make_v2_table; use crate::transaction::update_location::UpdateLocationAction; @@ -92,8 +93,9 @@ mod tests { assert_eq!(tx.actions.len(), 1); let action = (*tx.actions[0]) - .downcast_ref::<UpdateLocationAction>() - .unwrap(); + .downcast_ref::<ActionEntry<UpdateLocationAction>>() + .unwrap() + .action(); assert_eq!( action.location, diff --git a/crates/iceberg/src/transaction/update_properties.rs b/crates/iceberg/src/transaction/update_properties.rs index 240b451ef..3050805d5 100644 --- a/crates/iceberg/src/transaction/update_properties.rs +++ b/crates/iceberg/src/transaction/update_properties.rs @@ -16,12 +16,11 @@ // under the License. use std::collections::{HashMap, HashSet}; -use std::sync::Arc; use async_trait::async_trait; use crate::table::Table; -use crate::transaction::action::{ActionCommit, TransactionAction}; +use crate::transaction::action::{ActionCommit, StatelessAction}; use crate::{Error, ErrorKind, Result, TableUpdate}; /// A transactional action that updates or removes table properties @@ -30,6 +29,7 @@ use crate::{Error, ErrorKind, Result, TableUpdate}; /// properties during a transaction. It supports setting new values for existing keys /// or adding new keys, as well as removing existing keys. Each key can only be updated /// or removed in a single action, not both. +#[derive(Clone)] pub struct UpdatePropertiesAction { updates: HashMap<String, String>, removals: HashSet<String>, @@ -75,8 +75,8 @@ impl UpdatePropertiesAction { } #[async_trait] -impl TransactionAction for UpdatePropertiesAction { - async fn commit(self: Arc<Self>, _table: &Table) -> Result<ActionCommit> { +impl StatelessAction for UpdatePropertiesAction { + async fn commit(&self, _table: &Table) -> Result<ActionCommit> { if let Some(overlapping_key) = self.removals.iter().find(|k| self.updates.contains_key(*k)) { return Err(Error::new( @@ -105,7 +105,7 @@ mod tests { use as_any::Downcast; use crate::transaction::Transaction; - use crate::transaction::action::ApplyTransactionAction; + use crate::transaction::action::{ActionEntry, ApplyTransactionAction}; use crate::transaction::tests::make_v2_table; use crate::transaction::update_properties::UpdatePropertiesAction; @@ -123,8 +123,9 @@ mod tests { assert_eq!(tx.actions.len(), 1); let action = (*tx.actions[0]) - .downcast_ref::<UpdatePropertiesAction>() - .unwrap(); + .downcast_ref::<ActionEntry<UpdatePropertiesAction>>() + .unwrap() + .action(); assert_eq!( action.updates, HashMap::from([("a".to_string(), "b".to_string())]) diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index eee7f0c00..e5c3f6512 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -26,7 +26,7 @@ use crate::spec::{ StructType, Type, }; use crate::table::Table; -use crate::transaction::action::{ActionCommit, TransactionAction}; +use crate::transaction::action::{ActionCommit, StatelessAction}; use crate::{Error, ErrorKind, Result, TableRequirement, TableUpdate}; // Default ID for a new column. This will be re-assigned to a fresh ID at commit time. @@ -37,7 +37,7 @@ const DEFAULT_FIELD_ID: i32 = 0; /// Use helper constructors such as [`AddColumn::optional`] and [`AddColumn::required`], /// optionally combined with the builder's `parent` and `doc` setters via /// [`AddColumn::builder`], then pass the value to [`UpdateSchemaAction::add_column`]. -#[derive(TypedBuilder)] +#[derive(Clone, TypedBuilder)] pub struct AddColumn { #[builder(default = None, setter(strip_option, into))] parent: Option<String>, @@ -112,6 +112,7 @@ impl AddColumn { /// let tx = action.apply(tx).unwrap(); /// let table = tx.commit(&catalog).await.unwrap(); /// ``` +#[derive(Clone)] pub struct UpdateSchemaAction { additions: Vec<AddColumn>, deletes: Vec<String>, @@ -328,12 +329,12 @@ fn rebuild_field( } // --------------------------------------------------------------------------- -// TransactionAction implementation +// StatelessAction implementation // --------------------------------------------------------------------------- #[async_trait] -impl TransactionAction for UpdateSchemaAction { - async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { +impl StatelessAction for UpdateSchemaAction { + async fn commit(&self, table: &Table) -> Result<ActionCommit> { let base_schema = table.metadata().current_schema(); let mut last_column_id = table.metadata().last_column_id(); @@ -473,7 +474,6 @@ impl TransactionAction for UpdateSchemaAction { #[cfg(test)] mod tests { use std::io::BufReader; - use std::sync::Arc; use as_any::Downcast; @@ -482,8 +482,9 @@ mod tests { VariantType, }; use crate::table::Table; + use crate::test_utils::commit_with_fresh_state; use crate::transaction::Transaction; - use crate::transaction::action::{ApplyTransactionAction, TransactionAction}; + use crate::transaction::action::{ActionEntry, ApplyTransactionAction}; use crate::transaction::tests::make_v2_table; use crate::transaction::update_schema::{AddColumn, DEFAULT_FIELD_ID, UpdateSchemaAction}; use crate::{ErrorKind, TableIdent, TableRequirement, TableUpdate}; @@ -618,7 +619,7 @@ mod tests { Type::Primitive(PrimitiveType::Int), )); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let requirements = action_commit.take_requirements(); @@ -666,7 +667,7 @@ mod tests { .build(), ); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -693,7 +694,7 @@ mod tests { Literal::int(0), )); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -721,7 +722,7 @@ mod tests { Type::Primitive(PrimitiveType::Int), )); - let result = Arc::new(action).commit(&table).await; + let result = commit_with_fresh_state(action, &table).await; let err = match result { Err(e) => e, Ok(_) => panic!("should reject adding a column with an existing name"), @@ -742,7 +743,7 @@ mod tests { // z is not an identifier field, so we can delete it. let action = tx.update_schema().delete_column("z"); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -765,7 +766,7 @@ mod tests { let action = tx.update_schema().delete_column("nonexistent"); - let result = Arc::new(action).commit(&table).await; + let result = commit_with_fresh_state(action, &table).await; let err = match result { Err(e) => e, Ok(_) => panic!("should reject deleting a non-existent column"), @@ -792,7 +793,7 @@ mod tests { Type::Primitive(PrimitiveType::Boolean), )); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -823,7 +824,7 @@ mod tests { Type::Primitive(PrimitiveType::Boolean), )); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -854,7 +855,7 @@ mod tests { assert_eq!(tx.actions.len(), 1); (*tx.actions[0]) - .downcast_ref::<UpdateSchemaAction>() + .downcast_ref::<ActionEntry<UpdateSchemaAction>>() .expect("UpdateSchemaAction was not applied to Transaction!"); } @@ -876,7 +877,7 @@ mod tests { .build(), ); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -911,7 +912,7 @@ mod tests { .build(), ); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -941,7 +942,7 @@ mod tests { .build(), ); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -976,7 +977,7 @@ mod tests { .build(), ); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -1006,7 +1007,7 @@ mod tests { .build(), ); - let err = match Arc::new(action).commit(&table).await { + let err = match commit_with_fresh_state(action, &table).await { Err(e) => e, Ok(_) => panic!("should reject adding to a nonexistent parent"), }; @@ -1032,7 +1033,7 @@ mod tests { .build(), ); - let err = match Arc::new(action).commit(&table).await { + let err = match commit_with_fresh_state(action, &table).await { Err(e) => e, Ok(_) => panic!("should reject adding to a primitive parent"), }; @@ -1058,7 +1059,7 @@ mod tests { .build(), ); - let err = match Arc::new(action).commit(&table).await { + let err = match commit_with_fresh_state(action, &table).await { Err(e) => e, Ok(_) => panic!("should reject adding a column with conflicting name"), }; @@ -1090,7 +1091,7 @@ mod tests { .build(), ); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { @@ -1136,7 +1137,7 @@ mod tests { ])), )); - let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); let updates = action_commit.take_updates(); let new_schema = match &updates[0] { diff --git a/crates/iceberg/src/transaction/update_statistics.rs b/crates/iceberg/src/transaction/update_statistics.rs index e9fa86f10..f42765ed4 100644 --- a/crates/iceberg/src/transaction/update_statistics.rs +++ b/crates/iceberg/src/transaction/update_statistics.rs @@ -16,16 +16,16 @@ // under the License. use std::collections::HashMap; -use std::sync::Arc; use async_trait::async_trait; use crate::spec::StatisticsFile; use crate::table::Table; -use crate::transaction::{ActionCommit, TransactionAction}; +use crate::transaction::{ActionCommit, StatelessAction}; use crate::{Result, TableUpdate}; /// A transactional action for updating statistics files in a table +#[derive(Clone)] pub struct UpdateStatisticsAction { statistics_to_set: HashMap<i64, Option<StatisticsFile>>, } @@ -69,8 +69,8 @@ impl UpdateStatisticsAction { } #[async_trait] -impl TransactionAction for UpdateStatisticsAction { - async fn commit(self: Arc<Self>, _table: &Table) -> Result<ActionCommit> { +impl StatelessAction for UpdateStatisticsAction { + async fn commit(&self, _table: &Table) -> Result<ActionCommit> { let mut updates: Vec<TableUpdate> = vec![]; self.statistics_to_set @@ -98,6 +98,7 @@ mod tests { use as_any::Downcast; use crate::spec::{BlobMetadata, StatisticsFile}; + use crate::transaction::action::ActionEntry; use crate::transaction::tests::make_v2_table; use crate::transaction::update_statistics::UpdateStatisticsAction; use crate::transaction::{ApplyTransactionAction, Transaction}; @@ -147,8 +148,9 @@ mod tests { .unwrap(); let action = (*tx.actions[0]) - .downcast_ref::<UpdateStatisticsAction>() - .unwrap(); + .downcast_ref::<ActionEntry<UpdateStatisticsAction>>() + .unwrap() + .action(); assert!( action .statistics_to_set @@ -188,8 +190,9 @@ mod tests { .unwrap(); let action = (*tx.actions[0]) - .downcast_ref::<UpdateStatisticsAction>() - .unwrap(); + .downcast_ref::<ActionEntry<UpdateStatisticsAction>>() + .unwrap() + .action(); // Verify that the statistics file is set correctly assert_eq!( @@ -211,8 +214,9 @@ mod tests { let tx = tx.update_statistics().apply(tx).unwrap(); let action = (*tx.actions[0]) - .downcast_ref::<UpdateStatisticsAction>() - .unwrap(); + .downcast_ref::<ActionEntry<UpdateStatisticsAction>>() + .unwrap() + .action(); // Verify that no statistics are set assert!(action.statistics_to_set.is_empty()); diff --git a/crates/iceberg/src/transaction/upgrade_format_version.rs b/crates/iceberg/src/transaction/upgrade_format_version.rs index b833b5722..c65f955c6 100644 --- a/crates/iceberg/src/transaction/upgrade_format_version.rs +++ b/crates/iceberg/src/transaction/upgrade_format_version.rs @@ -15,8 +15,6 @@ // specific language governing permissions and limitations // under the License. -use std::sync::Arc; - use async_trait::async_trait; use crate::Result; @@ -24,13 +22,14 @@ use crate::TableUpdate::UpgradeFormatVersion; use crate::error::invalid_data; use crate::spec::FormatVersion; use crate::table::Table; -use crate::transaction::action::{ActionCommit, TransactionAction}; +use crate::transaction::action::{ActionCommit, StatelessAction}; /// A transaction action to upgrade a table's format version. /// /// This action is used within a transaction to indicate that the /// table's format version should be upgraded to a specified version. /// The location remains optional until explicitly set via [`UpgradeFormatVersionAction::set_format_version`]. +#[derive(Clone)] pub struct UpgradeFormatVersionAction { format_version: Option<FormatVersion>, } @@ -59,8 +58,8 @@ impl UpgradeFormatVersionAction { } #[async_trait] -impl TransactionAction for UpgradeFormatVersionAction { - async fn commit(self: Arc<Self>, _table: &Table) -> Result<ActionCommit> { +impl StatelessAction for UpgradeFormatVersionAction { + async fn commit(&self, _table: &Table) -> Result<ActionCommit> { let format_version = self.format_version.ok_or_else(|| { invalid_data!("FormatVersion is not set for UpgradeFormatVersionAction!") })?; @@ -78,7 +77,7 @@ mod tests { use crate::spec::FormatVersion; use crate::transaction::Transaction; - use crate::transaction::action::ApplyTransactionAction; + use crate::transaction::action::{ActionEntry, ApplyTransactionAction}; use crate::transaction::upgrade_format_version::UpgradeFormatVersionAction; #[test] @@ -94,8 +93,9 @@ mod tests { assert_eq!(tx.actions.len(), 1); let action = (*tx.actions[0]) - .downcast_ref::<UpgradeFormatVersionAction>() - .unwrap(); + .downcast_ref::<ActionEntry<UpgradeFormatVersionAction>>() + .unwrap() + .action(); assert_eq!(action.format_version, Some(FormatVersion::V2)); }
