laskoviymishka commented on code in PR #3265:
URL: https://github.com/apache/iceberg-rust/pull/3265#discussion_r4264337566
##########
crates/iceberg/src/transaction/action.rs:
##########
@@ -25,28 +24,175 @@ 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`]).
Review Comment:
nit: `Transaction` isn't `Clone` anymore, so this bit's just wrong now. I'd
drop it:
```suggestion
/// attempts of one logical execution but is never shared between executions.
```
##########
crates/iceberg/src/transaction/action.rs:
##########
@@ -25,28 +24,175 @@ 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 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) -> Self::State {}
+
+ 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 this action won't be retryable.
Review Comment:
nit: retry is a transaction-level thing, not an action property.
```suggestion
/// delete nothing, and the transaction must not be retried.
```
##########
crates/iceberg/src/transaction/action.rs:
##########
@@ -159,14 +349,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;
Review Comment:
This shows the unbox runs, not that state/status actually reach `cleanup`,
since the body's empty. Fine to leave for milestone 3 as you said. If you want
it now, a `CountingAction` that records `(state, status)` into a test-held sink
would do it. Not blocking.
--
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]
