blackmwk commented on code in PR #3265:
URL: https://github.com/apache/iceberg-rust/pull/3265#discussion_r4217114738
##########
crates/iceberg/src/transaction/mod.rs:
##########
@@ -84,10 +84,25 @@ 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 Clone for Transaction {
Review Comment:
Why we need this? If you want to retry a tx, we should change the table.
##########
crates/iceberg/src/transaction/action.rs:
##########
@@ -25,28 +24,147 @@ 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 ErasedActionEntry>;
/// 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
+/// (cloning a transaction creates fresh state via
[`TransactionAction::new_state`]).
#[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 use `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: the type system guarantees that no
+ /// further attempt can run for this entry after cleanup. 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)]
Review Comment:
Create a tracking issue to remove this attribute.
##########
crates/iceberg/src/transaction/action.rs:
##########
@@ -25,28 +24,147 @@ 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 ErasedActionEntry>;
/// 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
+/// (cloning a transaction creates fresh state via
[`TransactionAction::new_state`]).
#[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 use `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: the type system guarantees that no
+ /// further attempt can run for this entry after cleanup. 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: Box<Self>, state: 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.
+ ///
+ /// 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: Box<A>,
+ state: A::State,
+}
+
+impl<A: TransactionAction> ActionEntry<A> {
+ fn new(action: A) -> Self {
+ let state = action.new_state();
+ Self {
+ action: Box::new(action),
+ state,
+ }
+ }
+
+ /// The action (intent) held by this entry.
+ #[cfg(test)]
Review Comment:
I don't want to add this, I think it's fine to add this function.
##########
crates/iceberg/src/transaction/action.rs:
##########
@@ -25,28 +24,147 @@ 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 ErasedActionEntry>;
/// 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
+/// (cloning a transaction creates fresh state via
[`TransactionAction::new_state`]).
#[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 use `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: the type system guarantees that no
+ /// further attempt can run for this entry after cleanup. 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: Box<Self>, state: 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.
+ ///
+ /// 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: Box<A>,
+ state: A::State,
+}
+
+impl<A: TransactionAction> ActionEntry<A> {
+ fn new(action: A) -> Self {
+ let state = action.new_state();
+ Self {
+ action: Box::new(action),
+ state,
+ }
+ }
+
+ /// The action (intent) held by this entry.
+ #[cfg(test)]
+ 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 ErasedActionEntry: AsAny + Send + Sync {
Review Comment:
I'm not a big fine of this approach, it's odd to make `ErasedActionEntry` a
trait. Why not do following:
```
pub(crate) trait DynAction {
...
}
impl<T: Action> DynAction for T {
....
}
```
and store a Box<dyn DynAction> in ActionEntry?
##########
crates/iceberg/src/transaction/action.rs:
##########
@@ -103,26 +221,45 @@ impl ActionCommit {
}
}
+/// Runs a single commit attempt of an action with fresh state.
+///
+/// Test-only convenience mirroring how a transaction drives one attempt.
+#[cfg(test)]
+pub(crate) async fn commit_with_fresh_state<A: TransactionAction>(
Review Comment:
As we did in other mod, put the test util methods in test_util mod.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
