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;
19use std::sync::Arc;
20
21use as_any::AsAny;
22use async_trait::async_trait;
23
24use crate::table::Table;
25use crate::transaction::Transaction;
26use crate::{Result, TableRequirement, TableUpdate};
27
28/// A boxed, thread-safe reference to a `TransactionAction`.
29pub(crate) type BoxedTransactionAction = Arc<dyn TransactionAction>;
30
31/// A trait representing an atomic action that can be part of a transaction.
32///
33/// Implementors of this trait define how a specific action is committed to a table.
34/// Each action is responsible for generating the updates and requirements needed
35/// to modify the table metadata.
36#[async_trait]
37pub trait TransactionAction: AsAny + Sync + Send {
38    /// Commits this action against the provided table and returns the resulting updates.
39    /// NOTE: Most users should apply actions through [`Transaction`], which handles
40    /// rebasing onto the latest table state and retrying on commit conflicts. Call
41    /// this directly only to take over that responsibility yourself, passing the
42    /// resulting updates and requirements to [`Catalog::update_table`] via
43    /// [`TableCommit`].
44    ///
45    /// [`Catalog::update_table`]: crate::Catalog::update_table
46    /// [`TableCommit`]: crate::TableCommit
47    ///
48    /// # Arguments
49    ///
50    /// * `table` - The current state of the table this action should apply to.
51    ///
52    /// # Returns
53    ///
54    /// An `ActionCommit` containing table updates and table requirements,
55    /// or an error if the commit fails.
56    async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit>;
57}
58
59/// A helper trait for applying a `TransactionAction` to a `Transaction`.
60///
61/// This is implemented for all `TransactionAction` types
62/// to allow easy chaining of actions into a transaction context.
63pub trait ApplyTransactionAction {
64    /// Adds this action to the given transaction.
65    ///
66    /// # Arguments
67    ///
68    /// * `tx` - The transaction to apply the action to.
69    ///
70    /// # Returns
71    ///
72    /// The modified transaction containing this action, or an error if the operation fails.
73    fn apply(self, tx: Transaction) -> Result<Transaction>;
74}
75
76impl<T: TransactionAction + 'static> ApplyTransactionAction for T {
77    fn apply(self, mut tx: Transaction) -> Result<Transaction>
78    where Self: Sized {
79        tx.actions.push(Arc::new(self));
80        Ok(tx)
81    }
82}
83
84/// The result of committing a `TransactionAction`.
85///
86/// This struct contains the updates to apply to the table's metadata
87/// and any preconditions that must be satisfied before the update can be committed.
88pub struct ActionCommit {
89    updates: Vec<TableUpdate>,
90    requirements: Vec<TableRequirement>,
91}
92
93impl ActionCommit {
94    /// Creates a new `ActionCommit` from the given updates and requirements.
95    pub fn new(updates: Vec<TableUpdate>, requirements: Vec<TableRequirement>) -> Self {
96        Self {
97            updates,
98            requirements,
99        }
100    }
101
102    /// Consumes and returns the list of table updates.
103    pub fn take_updates(&mut self) -> Vec<TableUpdate> {
104        take(&mut self.updates)
105    }
106
107    /// Consumes and returns the list of table requirements.
108    pub fn take_requirements(&mut self) -> Vec<TableRequirement> {
109        take(&mut self.requirements)
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use std::str::FromStr;
116    use std::sync::Arc;
117
118    use as_any::Downcast;
119    use async_trait::async_trait;
120    use uuid::Uuid;
121
122    use crate::table::Table;
123    use crate::transaction::Transaction;
124    use crate::transaction::action::{ActionCommit, ApplyTransactionAction, TransactionAction};
125    use crate::transaction::tests::make_v2_table;
126    use crate::{Result, TableRequirement, TableUpdate};
127
128    struct TestAction;
129
130    #[async_trait]
131    impl TransactionAction for TestAction {
132        async fn commit(self: Arc<Self>, _table: &Table) -> Result<ActionCommit> {
133            Ok(ActionCommit::new(
134                vec![TableUpdate::SetLocation {
135                    location: String::from("s3://bucket/prefix/table/"),
136                }],
137                vec![TableRequirement::UuidMatch {
138                    uuid: Uuid::from_str("9c12d441-03fe-4693-9a96-a0705ddf69c1")?,
139                }],
140            ))
141        }
142    }
143
144    #[tokio::test]
145    async fn test_commit_transaction_action() {
146        let table = make_v2_table();
147        let action = TestAction;
148
149        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
150
151        let updates = action_commit.take_updates();
152        let requirements = action_commit.take_requirements();
153
154        assert_eq!(updates[0], TableUpdate::SetLocation {
155            location: String::from("s3://bucket/prefix/table/")
156        });
157        assert_eq!(requirements[0], TableRequirement::UuidMatch {
158            uuid: Uuid::from_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap()
159        });
160    }
161
162    #[test]
163    fn test_apply_transaction_action() {
164        let table = make_v2_table();
165        let action = TestAction;
166        let tx = Transaction::new(&table);
167
168        let updated_tx = action.apply(tx).unwrap();
169        // There should be one action in the transaction now
170        assert_eq!(updated_tx.actions.len(), 1);
171
172        (*updated_tx.actions[0])
173            .downcast_ref::<TestAction>()
174            .expect("TestAction was not applied to Transaction!");
175    }
176
177    #[test]
178    fn test_action_commit() {
179        // Create dummy updates and requirements
180        let location = String::from("s3://bucket/prefix/table/");
181        let uuid = Uuid::new_v4();
182        let updates = vec![TableUpdate::SetLocation { location }];
183        let requirements = vec![TableRequirement::UuidMatch { uuid }];
184
185        let mut action_commit = ActionCommit::new(updates.clone(), requirements.clone());
186
187        let taken_updates = action_commit.take_updates();
188        let taken_requirements = action_commit.take_requirements();
189
190        // Check values are returned correctly
191        assert_eq!(taken_updates, updates);
192        assert_eq!(taken_requirements, requirements);
193
194        assert!(action_commit.take_updates().is_empty());
195        assert!(action_commit.take_requirements().is_empty());
196    }
197}
iceberg/transaction/action.rs

iceberg/transaction/
action.rs
