allow specify snapshot id for fast append (#25)(#14)

Signed-off-by: xxchan <[email protected]>

Author: xxchan

crates/iceberg/src/transaction/mod.rs

@@ -129,10 +129,26 @@ impl<'a> Transaction<'a> {
     /// Creates a fast append action.
     pub fn fast_append(
         self,
+        snapshot_id: Option<i64>,
         commit_uuid: Option<Uuid>,
         key_metadata: Vec<u8>,
     ) -> Result<FastAppendAction<'a>> {
-        let snapshot_id = self.generate_unique_snapshot_id();
+        let snapshot_id = if let Some(snapshot_id) = snapshot_id {
+            if self
+                .table
+                .metadata()
+                .snapshots()
+                .any(|s| s.snapshot_id() == snapshot_id)
+            {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    format!("Snapshot id {} already exists", snapshot_id),
+                ));
+            }
+            snapshot_id
+        } else {
+            self.generate_unique_snapshot_id()
+        };
         FastAppendAction::new(
             self,
             snapshot_id,

crates/integration_tests/tests/shared_tests/append_data_file_test.rs

@@ -112,7 +112,7 @@ async fn test_append_data_file() {
 
     // commit result
     let tx = Transaction::new(&table);
-    let mut append_action = tx.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx.fast_append(None, None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx = append_action.apply().await.unwrap();
     let table = tx.commit(&rest_catalog).await.unwrap();
@@ -132,7 +132,7 @@ async fn test_append_data_file() {
 
     // commit result again
     let tx = Transaction::new(&table);
-    let mut append_action = tx.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx.fast_append(None, None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx = append_action.apply().await.unwrap();
     let table = tx.commit(&rest_catalog).await.unwrap();

crates/integration_tests/tests/shared_tests/append_partition_data_file_test.rs

@@ -120,7 +120,7 @@ async fn test_append_partition_data_file() {
 
     // commit result
     let tx = Transaction::new(&table);
-    let mut append_action = tx.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx.fast_append(None, None, vec![]).unwrap();
     append_action
         .add_data_files(data_file_valid.clone())
         .unwrap();
@@ -180,7 +180,7 @@ async fn test_schema_incompatible_partition_type(
     let data_file_invalid = data_file_writer_invalid.close().await.unwrap();
 
     let tx = Transaction::new(&table);
-    let mut append_action = tx.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx.fast_append(None, None, vec![]).unwrap();
     if append_action
         .add_data_files(data_file_invalid.clone())
         .is_ok()
@@ -220,7 +220,7 @@ async fn test_schema_incompatible_partition_fields(
     let data_file_invalid = data_file_writer_invalid.close().await.unwrap();
 
     let tx = Transaction::new(&table);
-    let mut append_action = tx.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx.fast_append(None, None, vec![]).unwrap();
     if append_action
         .add_data_files(data_file_invalid.clone())
         .is_ok()

crates/integration_tests/tests/shared_tests/conflict_commit_test.rs

@@ -90,12 +90,12 @@ async fn test_append_data_file_conflict() {
 
     // start two transaction and commit one of them
     let tx1 = Transaction::new(&table);
-    let mut append_action = tx1.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx1.fast_append(None, None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx1 = append_action.apply().await.unwrap();
 
     let tx2 = Transaction::new(&table);
-    let mut append_action = tx2.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx2.fast_append(None, None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx2 = append_action.apply().await.unwrap();
     let table = tx2

crates/integration_tests/tests/shared_tests/scan_all_type.rs

@@ -309,7 +309,7 @@ async fn test_scan_all_type() {
 
     // commit result
     let tx = Transaction::new(&table);
-    let mut append_action = tx.fast_append(None, vec![]).unwrap();
+    let mut append_action = tx.fast_append(None, None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx = append_action.apply().await.unwrap();
     let table = tx.commit(&rest_catalog).await.unwrap();