iceberg/transaction/
action.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::mem::take;
19
20use as_any::AsAny;
21use async_trait::async_trait;
22
23use crate::table::Table;
24use crate::transaction::Transaction;
25use crate::{Result, TableRequirement, TableUpdate};
26
27/// A boxed entry pairing a transaction action with its retry-persistent state.
28pub(crate) type TransactionActionEntry = Box<dyn DynActionEntry>;
29
30/// A trait representing an atomic action that can be part of a transaction.
31///
32/// Implementors of this trait define how a specific action is committed to a table.
33/// Each action is responsible for generating the updates and requirements needed
34/// to modify the table metadata.
35///
36/// An action's intent is immutable once applied to a transaction. Retry-persistent
37/// state lives in the associated [`TransactionAction::State`], which survives replay
38/// attempts of one logical execution but is never shared between executions.
39#[async_trait]
40pub(crate) trait TransactionAction: Clone + Send + Sync + 'static {
41    /// Retry-persistent state exclusively owned by one logical execution of this
42    /// action. Stateless actions implement [`StatelessAction`] instead, which
43    /// supplies `State = ()`.
44    type State: Send + Sync + 'static;
45
46    /// Creates fresh state for one logical execution.
47    ///
48    /// This is infallible and table-independent; table-dependent initialization
49    /// happens during [`TransactionAction::commit`].
50    fn new_state(&self) -> Self::State;
51
52    /// Commits this action against the provided table and returns the resulting updates.
53    /// NOTE: This function is intended for internal use only and should not be called directly by users.
54    ///
55    /// One replay attempt: the action (intent) is borrowed immutably, its execution
56    /// state mutably, and the table reflects the current transaction-local base.
57    ///
58    /// # Arguments
59    ///
60    /// * `state` - The retry-persistent state for this logical execution.
61    /// * `table` - The current state of the table this action should apply to.
62    ///
63    /// # Returns
64    ///
65    /// An `ActionCommit` containing table updates and table requirements,
66    /// or an error if the commit fails.
67    async fn commit(&self, state: &mut Self::State, table: &Table) -> Result<ActionCommit>;
68
69    /// Best-effort cleanup after the transaction reaches a terminal result.
70    ///
71    /// Consumes the action and its state: ownership makes any further use
72    /// of this action after cleanup a compile error.
73    /// Cleanup must not change the already-determined transaction result.
74    ///
75    /// See [`CommitStatus`] for what each terminal status allows cleanup to do.
76    // TODO: invoke this from `Transaction::commit` once terminal status
77    // classification is wired up (stateful transaction RFC, milestone 3).
78    #[allow(dead_code)]
79    async fn cleanup(self, state: Self::State, table: &Table, status: CommitStatus);
80}
81
82/// Convenience trait for actions that carry no per-attempt state.
83///
84/// Implementing `StatelessAction` yields a [`TransactionAction`] impl with
85/// `State = ()`, a no-op [`TransactionAction::new_state`], a no-op
86/// [`TransactionAction::cleanup`], and a [`TransactionAction::commit`] that
87/// forwards to [`StatelessAction::commit`]. Actions that need retry-persistent
88/// state (e.g. a persistent snapshot producer) implement `TransactionAction`
89/// directly instead.
90#[async_trait]
91pub(crate) trait StatelessAction: Clone + Send + Sync + 'static {
92    /// One commit attempt against the current transaction-local table.
93    ///
94    /// Same contract as [`TransactionAction::commit`], minus the state parameter.
95    async fn commit(&self, table: &Table) -> Result<ActionCommit>;
96}
97
98#[async_trait]
99impl<T: StatelessAction> TransactionAction for T {
100    type State = ();
101
102    fn new_state(&self) {}
103
104    async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> {
105        StatelessAction::commit(self, table).await
106    }
107
108    async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {}
109}
110
111/// Classification of a transaction's terminal result, consumed by
112/// [`TransactionAction::cleanup`] to determine what is safe to delete.
113///
114/// The transaction/catalog layer determines this classification; actions
115/// consume it rather than independently interpreting catalog errors.
116// TODO: produce this classification in `Transaction::commit` once terminal
117// cleanup is wired up (stateful transaction RFC, milestone 3).
118#[allow(dead_code)]
119#[derive(Debug, Clone, Copy, PartialEq, Eq)]
120pub(crate) enum CommitStatus {
121    /// The catalog confirmed that the commit was applied. Cleanup may remove
122    /// owned artifacts not retained by the committed result.
123    Committed,
124    /// The transaction definitively did not commit and will not retry.
125    /// Owned artifacts are deletable.
126    Failed,
127    /// The commit request was submitted, but its outcome could not be
128    /// resolved: the catalog may or may not have applied it. Cleanup must
129    /// delete nothing, and the transaction must not be retried.
130    ///
131    /// Example: every action executed and validated successfully, but the
132    /// connection failed while awaiting the catalog's response to
133    /// `update_table`.
134    Unknown,
135}
136
137/// An entry in a transaction, pairing an action's immutable intent with the
138/// retry-persistent state of one logical execution.
139///
140/// The pairing is preserved by construction: the entry is created with fresh
141/// state and owns both exclusively, so terminal cleanup can consume them together.
142pub(crate) struct ActionEntry<A: TransactionAction> {
143    action: A,
144    state: A::State,
145}
146
147impl<A: TransactionAction> ActionEntry<A> {
148    fn new(action: A) -> Self {
149        let state = action.new_state();
150        Self { action, state }
151    }
152
153    /// The action (intent) held by this entry.
154    #[allow(dead_code)]
155    pub(crate) fn action(&self) -> &A {
156        &self.action
157    }
158}
159
160/// Object-safe adapter over [`ActionEntry`], allowing a transaction to store
161/// heterogeneous entries while keeping each action paired with its own state.
162#[async_trait]
163pub(crate) trait DynActionEntry: AsAny + Send + Sync {
164    /// One replay attempt against the current transaction-local table.
165    async fn commit(&mut self, table: &Table) -> Result<ActionCommit>;
166
167    /// Best-effort terminal cleanup, consuming the entry.
168    // TODO: invoke from `Transaction::commit` once terminal status
169    // classification is wired up (stateful transaction RFC, milestone 3).
170    #[allow(dead_code)]
171    async fn cleanup(self: Box<Self>, table: &Table, status: CommitStatus);
172
173    /// Creates a new entry for a new logical execution of the same action:
174    /// the action (intent) is cloned, and it is paired with fresh state from
175    /// [`TransactionAction::new_state`]. Accrued retry state is never carried
176    /// over to the new entry.
177    #[allow(dead_code)]
178    fn clone_with_new_state(&self) -> TransactionActionEntry;
179}
180
181#[async_trait]
182impl<A: TransactionAction> DynActionEntry for ActionEntry<A> {
183    async fn commit(&mut self, table: &Table) -> Result<ActionCommit> {
184        self.action.commit(&mut self.state, table).await
185    }
186
187    async fn cleanup(self: Box<Self>, table: &Table, status: CommitStatus) {
188        let ActionEntry { action, state } = *self;
189        action.cleanup(state, table, status).await
190    }
191
192    fn clone_with_new_state(&self) -> TransactionActionEntry {
193        Box::new(ActionEntry::new(self.action.clone()))
194    }
195}
196
197/// A helper trait for applying a `TransactionAction` to a `Transaction`.
198///
199/// This is implemented for all `TransactionAction` types
200/// to allow easy chaining of actions into a transaction context.
201pub trait ApplyTransactionAction {
202    /// Adds this action to the given transaction.
203    ///
204    /// # Arguments
205    ///
206    /// * `tx` - The transaction to apply the action to.
207    ///
208    /// # Returns
209    ///
210    /// The modified transaction containing this action, or an error if the operation fails.
211    fn apply(self, tx: Transaction) -> Result<Transaction>;
212}
213
214impl<T: TransactionAction> ApplyTransactionAction for T {
215    fn apply(self, mut tx: Transaction) -> Result<Transaction>
216    where Self: Sized {
217        tx.actions.push(Box::new(ActionEntry::new(self)));
218        Ok(tx)
219    }
220}
221
222/// The result of committing a `TransactionAction`.
223///
224/// This struct contains the updates to apply to the table's metadata
225/// and any preconditions that must be satisfied before the update can be committed.
226pub struct ActionCommit {
227    updates: Vec<TableUpdate>,
228    requirements: Vec<TableRequirement>,
229}
230
231impl ActionCommit {
232    /// Creates a new `ActionCommit` from the given updates and requirements.
233    pub fn new(updates: Vec<TableUpdate>, requirements: Vec<TableRequirement>) -> Self {
234        Self {
235            updates,
236            requirements,
237        }
238    }
239
240    /// Consumes and returns the list of table updates.
241    pub fn take_updates(&mut self) -> Vec<TableUpdate> {
242        take(&mut self.updates)
243    }
244
245    /// Consumes and returns the list of table requirements.
246    pub fn take_requirements(&mut self) -> Vec<TableRequirement> {
247        take(&mut self.requirements)
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use std::str::FromStr;
254
255    use as_any::Downcast;
256    use async_trait::async_trait;
257    use uuid::Uuid;
258
259    use crate::table::Table;
260    use crate::test_utils::commit_with_fresh_state;
261    use crate::transaction::Transaction;
262    use crate::transaction::action::{
263        ActionCommit, ActionEntry, ApplyTransactionAction, CommitStatus, DynActionEntry,
264        StatelessAction, TransactionAction,
265    };
266    use crate::transaction::tests::make_v2_table;
267    use crate::{Result, TableRequirement, TableUpdate};
268
269    #[derive(Clone)]
270    struct TestAction;
271
272    #[async_trait]
273    impl TransactionAction for TestAction {
274        type State = ();
275
276        fn new_state(&self) -> Self::State {}
277
278        async fn commit(&self, _state: &mut (), _table: &Table) -> Result<ActionCommit> {
279            Ok(ActionCommit::new(
280                vec![TableUpdate::SetLocation {
281                    location: String::from("s3://bucket/prefix/table/"),
282                }],
283                vec![TableRequirement::UuidMatch {
284                    uuid: Uuid::from_str("9c12d441-03fe-4693-9a96-a0705ddf69c1")?,
285                }],
286            ))
287        }
288
289        async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {}
290    }
291
292    #[derive(Clone)]
293    struct CountingAction;
294
295    #[async_trait]
296    impl TransactionAction for CountingAction {
297        type State = u32;
298
299        fn new_state(&self) -> Self::State {
300            0
301        }
302
303        async fn commit(&self, state: &mut u32, _table: &Table) -> Result<ActionCommit> {
304            *state += 1;
305            Ok(ActionCommit::new(vec![], vec![]))
306        }
307
308        async fn cleanup(self, _state: u32, _table: &Table, _status: CommitStatus) {}
309    }
310
311    #[derive(Clone)]
312    struct StatelessTestAction;
313
314    #[async_trait]
315    impl StatelessAction for StatelessTestAction {
316        async fn commit(&self, _table: &Table) -> Result<ActionCommit> {
317            Ok(ActionCommit::new(
318                vec![TableUpdate::SetLocation {
319                    location: String::from("s3://bucket/prefix/stateless/"),
320                }],
321                vec![],
322            ))
323        }
324    }
325
326    #[tokio::test]
327    async fn test_commit_transaction_action() {
328        let table = make_v2_table();
329        let action = TestAction;
330
331        let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap();
332
333        let updates = action_commit.take_updates();
334        let requirements = action_commit.take_requirements();
335
336        assert_eq!(updates[0], TableUpdate::SetLocation {
337            location: String::from("s3://bucket/prefix/table/")
338        });
339        assert_eq!(requirements[0], TableRequirement::UuidMatch {
340            uuid: Uuid::from_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap()
341        });
342    }
343
344    #[test]
345    fn test_apply_transaction_action() {
346        let table = make_v2_table();
347        let action = TestAction;
348        let tx = Transaction::new(&table);
349
350        let updated_tx = action.apply(tx).unwrap();
351        // There should be one action entry in the transaction now
352        assert_eq!(updated_tx.actions.len(), 1);
353
354        (*updated_tx.actions[0])
355            .downcast_ref::<ActionEntry<TestAction>>()
356            .expect("TestAction was not applied to Transaction!");
357    }
358
359    #[tokio::test]
360    async fn test_entry_retains_state_across_commits() {
361        let table = make_v2_table();
362        let mut tx = CountingAction.apply(Transaction::new(&table)).unwrap();
363
364        let entry = &mut tx.actions[0];
365        entry.commit(&table).await.unwrap();
366        entry.commit(&table).await.unwrap();
367
368        let entry = (**entry)
369            .downcast_ref::<ActionEntry<CountingAction>>()
370            .unwrap();
371        assert_eq!(entry.state, 2);
372    }
373
374    #[tokio::test]
375    async fn test_clone_with_new_state_resets_state() {
376        let table = make_v2_table();
377        let mut tx = CountingAction.apply(Transaction::new(&table)).unwrap();
378        tx.actions[0].commit(&table).await.unwrap();
379
380        let cloned = tx.actions[0].clone_with_new_state();
381
382        let original = (*tx.actions[0])
383            .downcast_ref::<ActionEntry<CountingAction>>()
384            .unwrap();
385        let cloned = (*cloned)
386            .downcast_ref::<ActionEntry<CountingAction>>()
387            .unwrap();
388        assert_eq!(original.state, 1);
389        assert_eq!(cloned.state, 0);
390    }
391
392    #[tokio::test]
393    async fn test_stateless_action_blanket_impl() {
394        let table = make_v2_table();
395
396        // commit goes through the blanket TransactionAction impl
397        let mut commit = commit_with_fresh_state(StatelessTestAction, &table)
398            .await
399            .unwrap();
400        assert_eq!(commit.take_updates(), vec![TableUpdate::SetLocation {
401            location: String::from("s3://bucket/prefix/stateless/"),
402        }]);
403
404        // ApplyTransactionAction is available through the blanket impl too
405        let tx = StatelessTestAction.apply(Transaction::new(&table)).unwrap();
406        assert_eq!(tx.actions.len(), 1);
407
408        // and the erased cleanup forwarding chain compiles and runs
409        let entry: Box<ActionEntry<StatelessTestAction>> =
410            Box::new(ActionEntry::new(StatelessTestAction));
411        entry.cleanup(&table, CommitStatus::Committed).await;
412    }
413
414    #[test]
415    fn test_action_commit() {
416        // Create dummy updates and requirements
417        let location = String::from("s3://bucket/prefix/table/");
418        let uuid = Uuid::new_v4();
419        let updates = vec![TableUpdate::SetLocation { location }];
420        let requirements = vec![TableRequirement::UuidMatch { uuid }];
421
422        let mut action_commit = ActionCommit::new(updates.clone(), requirements.clone());
423
424        let taken_updates = action_commit.take_updates();
425        let taken_requirements = action_commit.take_requirements();
426
427        // Check values are returned correctly
428        assert_eq!(taken_updates, updates);
429        assert_eq!(taken_requirements, requirements);
430
431        assert!(action_commit.take_updates().is_empty());
432        assert!(action_commit.take_requirements().is_empty());
433    }
434}
iceberg/transaction/action.rs

iceberg/transaction/
action.rs
