feat(transaction): Add TransactionAction POC

crates/catalog/memory/src/catalog.rs

@@ -283,6 +283,44 @@ impl Catalog for MemoryCatalog {
             "MemoryCatalog does not currently support updating tables.",
         ))
     }
+
+    // async fn commit_table(&self, base: &Table, current: Table) -> Result<Table> {
+    //     if base.metadata() == current.metadata() {
+    //         // no change
+    //         return Ok(current);
+    //     }
+    //
+    //     let mut root_namespace_state = self.root_namespace_state.lock().await;
+    //     // TODO: caller needs to retry on the error below
+    //     let _ = root_namespace_state
+    //         .check_metadata_location(base.identifier(), base.metadata_location())?;
+    //
+    //     let next_metadata_version = if let Some(base_metadata_location) = base.metadata_location() {
+    //         self.parse_metadata_version(base_metadata_location) + 1
+    //     } else {
+    //         0
+    //     };
+    //
+    //     // write metadata
+    //     let metadata_location = format!(
+    //         "{}/metadata/{}-{}.metadata.json",
+    //         current.metadata().location(),
+    //         next_metadata_version,
+    //         Uuid::new_v4()
+    //     );
+    //
+    //     // TODO instead of using current.metadata(), build a new metadata with some properties like last_updated_ms updated
+    //     self.file_io
+    //         .new_output(&metadata_location)?
+    //         .write(serde_json::to_vec(current.metadata())?.into())
+    //         .await?;
+    //
+    //     root_namespace_state
+    //         .update_existing_table_location(current.identifier(), current.metadata_location())?;
+    //
+    //     // TODO same here, need to update the metadata location
+    //     Ok(current)
+    // }
 }
 
 #[cfg(test)]

crates/catalog/memory/src/namespace_state.rs

@@ -262,6 +262,45 @@ impl NamespaceState {
         }
     }
 
+    pub(crate) fn check_metadata_location(
+        &self,
+        table_ident: &TableIdent,
+        metadata_location: Option<&str>,
+    ) -> Result<()> {
+        let namespace = self.get_namespace(table_ident.namespace())?;
+
+        if namespace
+            .table_metadata_locations
+            .get(table_ident.name())
+            .map(|s| s.as_str())
+            != metadata_location
+        {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!("Metadata location does not match for table: {table_ident}!"),
+            ));
+        }
+
+        Ok(())
+    }
+
+    pub(crate) fn update_existing_table_location(
+        &mut self,
+        table_ident: &TableIdent,
+        new_metadata_location: Option<&str>,
+    ) -> Result<()> {
+        if new_metadata_location.is_none() {
+            return Ok(());
+        }
+
+        let namespace = self.get_mut_namespace(table_ident.namespace())?;
+        namespace
+            .table_metadata_locations
+            .entry(table_ident.name().to_string())
+            .insert_entry(new_metadata_location.unwrap().to_string());
+        Ok(())
+    }
+
     // Inserts the given table or returns an error if it already exists
     pub(crate) fn insert_new_table(
         &mut self,

crates/catalog/rest/src/catalog.rs

@@ -2125,10 +2125,10 @@ mod tests {
                 .unwrap()
         };
 
-        let table = Transaction::new(&table1)
+        let table = Transaction::new(table1)
             .upgrade_table_version(FormatVersion::V2)
             .unwrap()
-            .commit(&catalog)
+            .commit(Arc::new(&catalog))
             .await
             .unwrap();
 
@@ -2250,10 +2250,10 @@ mod tests {
                 .unwrap()
         };
 
-        let table_result = Transaction::new(&table1)
+        let table_result = Transaction::new(table1)
             .upgrade_table_version(FormatVersion::V2)
             .unwrap()
-            .commit(&catalog)
+            .commit(Arc::new(&catalog))
             .await;
 
         assert!(table_result.is_err());

crates/catalog/rest/tests/rest_catalog_test.rs

@@ -19,7 +19,7 @@
 
 use std::collections::HashMap;
 use std::net::SocketAddr;
-use std::sync::RwLock;
+use std::sync::{Arc, RwLock};
 
 use ctor::{ctor, dtor};
 use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
@@ -347,10 +347,10 @@ async fn test_update_table() {
     );
 
     // Update table by committing transaction
-    let table2 = Transaction::new(&table)
+    let table2 = Transaction::new(table)
         .set_properties(HashMap::from([("prop1".to_string(), "v1".to_string())]))
         .unwrap()
-        .commit(&catalog)
+        .commit(Arc::new(&catalog))
         .await
         .unwrap();
 

crates/iceberg/Cargo.toml

@@ -55,6 +55,7 @@ arrow-select = { workspace = true }
 arrow-string = { workspace = true }
 async-std = { workspace = true, optional = true, features = ["attributes"] }
 async-trait = { workspace = true }
+backon = { version = "1.5.1"}
 base64 = { workspace = true }
 bimap = { workspace = true }
 bytes = { workspace = true }
@@ -86,6 +87,7 @@ typed-builder = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
 zstd = { workspace = true }
+regex = "1.11.1"
 
 [dev-dependencies]
 ctor = { workspace = true }

crates/iceberg/src/catalog/mod.rs

@@ -21,6 +21,7 @@ use std::collections::HashMap;
 use std::fmt::{Debug, Display};
 use std::mem::take;
 use std::ops::Deref;
+use std::sync::Arc;
 
 use _serde::deserialize_snapshot;
 use async_trait::async_trait;
@@ -300,6 +301,27 @@ impl TableCommit {
     pub fn take_updates(&mut self) -> Vec<TableUpdate> {
         take(&mut self.updates)
     }
+
+    /// Apply updates to a table
+    pub fn apply(&mut self, mut table: Table) -> Result<Table> {
+        // 1. check requirements
+        let requirements = self.take_requirements();
+        for requirement in requirements {
+            requirement.check(Some(table.metadata()))?;
+        }
+
+        // 2. Apply updates to metadata builder
+        let mut metadata_builder = table.metadata().clone().into_builder(None);
+
+        let updates = self.take_updates();
+        for update in updates {
+            metadata_builder = update.apply(metadata_builder)?;
+        }
+
+        table.with_metadata(Arc::new(metadata_builder.build()?.metadata));
+
+        Ok(table)
+    }
 }
 
 /// TableRequirement represents a requirement for a table in the catalog.

crates/iceberg/src/error.rs

@@ -39,7 +39,7 @@ pub enum ErrorKind {
     /// Iceberg data is invalid.
     ///
     /// This error is returned when we try to read a table from iceberg but
-    /// failed to parse it's metadata or data file correctly.
+    /// failed to parse its metadata or data file correctly.
     ///
     /// The table could be invalid or corrupted.
     DataInvalid,

crates/iceberg/src/spec/table_metadata.rs

@@ -26,6 +26,7 @@ use std::sync::Arc;
 
 use _serde::TableMetadataEnum;
 use chrono::{DateTime, Utc};
+use regex::Regex;
 use serde::{Deserialize, Serialize};
 use serde_repr::{Deserialize_repr, Serialize_repr};
 use uuid::Uuid;
@@ -40,6 +41,7 @@ use crate::error::{Result, timestamp_ms_to_utc};
 use crate::{Error, ErrorKind};
 
 static MAIN_BRANCH: &str = "main";
+const TABLE_METADATA_FILE_NAME_REGEX: &str = r"(\d+)-([\w-]{36})(?:\.\w+)?\.metadata\.json";
 pub(crate) static ONE_MINUTE_MS: i64 = 60_000;
 
 pub(crate) static EMPTY_SNAPSHOT_ID: i64 = -1;
@@ -432,6 +434,30 @@ impl TableMetadata {
         self.encryption_keys.get(key_id)
     }
 
+    /// Parse metadata version and uuid from metadata filename
+    fn parse_metadata_filename(metadata_location: &str) -> Result<(i32, Uuid)> {
+        if let Some(metadata_file_name) = metadata_location.split('/').last() {
+            let re = Regex::new(TABLE_METADATA_FILE_NAME_REGEX)
+                .expect("Failed to parse regex for metadata file!");
+            if let Some(caps) = re.captures(metadata_file_name) {
+                let metadata_version_str = &caps[1];
+                let uuid_str = &caps[2];
+
+                let metadata_version = metadata_version_str
+                    .parse()
+                    .expect(format!("Invalid metadata version: {metadata_version_str}").as_str());
+                let uuid = Uuid::parse_str(uuid_str)?;
+
+                return Ok((metadata_version, uuid));
+            }
+        }
+
+        Err(Error::new(
+            ErrorKind::Unexpected,
+            format!("Unrecognizable metadata location: {metadata_location}"),
+        ))
+    }
+
     /// Normalize this partition spec.
     ///
     /// This is an internal method

crates/iceberg/src/transaction/action/mod.rs

@@ -0,0 +1,73 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::any::Any;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+
+use crate::transaction::Transaction;
+use crate::{Error, ErrorKind, Result, TableRequirement, TableUpdate};
+
+pub type BoxedTransactionAction = Arc<dyn TransactionAction>;
+
+#[async_trait]
+pub(crate) trait TransactionAction: Sync + Send {
+    fn apply(&self) -> Box<dyn Any>;
+
+    /// Commit the changes and apply the changes to the transaction
+    fn commit(self: Arc<Self>, tx: &mut Transaction) -> Result<()>;
+}
+
+pub struct UpdateLocationAction {
+    location: Option<String>,
+}
+
+impl UpdateLocationAction {
+    pub fn new() -> Self {
+        UpdateLocationAction { location: None }
+    }
+
+    pub fn set_location(mut self, location: String) -> Self {
+        self.location = Some(location);
+        self
+    }
+}
+
+impl TransactionAction for UpdateLocationAction {
+    fn apply(&self) -> Box<dyn Any> {
+        Box::new(self.location.clone())
+    }
+
+    fn commit(self: Arc<Self>, tx: &mut Transaction) -> Result<()> {
+        let updates: Vec<TableUpdate>;
+        let requirements: Vec<TableRequirement>;
+        if let Some(location) = self.location.clone() {
+            updates = vec![TableUpdate::SetLocation { location }];
+            requirements = vec![];
+        } else {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "Location is not set for SetLocation!",
+            ));
+        }
+
+        tx.actions.push(self);
+
+        tx.apply(updates, requirements)
+    }
+}