feat(transaction): Add atomic snapshot tagging to FastAppendAction

Add the ability to create a tag reference atomically in the same
transaction that creates a new snapshot via fast_append.

This adds a `with_tag()` method to `FastAppendAction` that allows
specifying a tag name. When set, the tag will be created pointing
to the newly created snapshot, all within a single atomic catalog
update.

Example usage:
```rust
let tx = Transaction::new(&table);
let action = tx
    .fast_append()
    .add_data_files(data_files)
    .with_tag("v1.0.0");
let tx = action.apply(tx)?;
let table = tx.commit(&catalog).await?;
```

Author: alexjbuck

crates/iceberg/src/transaction/append.rs

@@ -37,6 +37,8 @@ pub struct FastAppendAction {
     key_metadata: Option<Vec<u8>>,
     snapshot_properties: HashMap<String, String>,
     added_data_files: Vec<DataFile>,
+    // Optional tag name to create atomically with the snapshot.
+    tag_ref: Option<String>,
 }
 
 impl FastAppendAction {
@@ -47,6 +49,7 @@ impl FastAppendAction {
             key_metadata: None,
             snapshot_properties: HashMap::default(),
             added_data_files: vec![],
+            tag_ref: None,
         }
     }
 
@@ -79,6 +82,12 @@ impl FastAppendAction {
         self.snapshot_properties = snapshot_properties;
         self
     }
+
+    /// Set a tag name to be created atomically with the snapshot.
+    pub fn with_tag(mut self, tag_name: impl Into<String>) -> Self {
+        self.tag_ref = Some(tag_name.into());
+        self
+    }
 }
 
 #[async_trait]
@@ -90,6 +99,7 @@ impl TransactionAction for FastAppendAction {
             self.key_metadata.clone(),
             self.snapshot_properties.clone(),
             self.added_data_files.clone(),
+            self.tag_ref.clone(),
         );
 
         // validate added files
@@ -333,4 +343,65 @@ mod tests {
         );
         assert_eq!(data_file, *manifest.entries()[0].data_file());
     }
+
+    #[tokio::test]
+    async fn test_fast_append_with_tag() {
+        let table = make_v2_minimal_table();
+        let tx = Transaction::new(&table);
+
+        let data_file = DataFileBuilder::default()
+            .content(DataContentType::Data)
+            .file_path("test/tagged.parquet".to_string())
+            .file_format(DataFileFormat::Parquet)
+            .file_size_in_bytes(100)
+            .record_count(1)
+            .partition_spec_id(table.metadata().default_partition_spec_id())
+            .partition(Struct::from_iter([Some(Literal::long(300))]))
+            .build()
+            .unwrap();
+
+        let action = tx
+            .fast_append()
+            .add_data_files(vec![data_file])
+            .with_tag("v1.0.0");
+
+        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
+        let updates = action_commit.take_updates();
+
+        // Should have 3 updates: AddSnapshot, SetSnapshotRef (main), SetSnapshotRef (tag)
+        assert_eq!(updates.len(), 3);
+
+        // First update: AddSnapshot
+        let snapshot_id = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
+            snapshot.snapshot_id()
+        } else {
+            panic!("Expected AddSnapshot as first update");
+        };
+
+        // Second update: SetSnapshotRef for main branch
+        if let TableUpdate::SetSnapshotRef {
+            ref_name,
+            reference,
+        } = &updates[1]
+        {
+            assert_eq!(ref_name, MAIN_BRANCH);
+            assert_eq!(reference.snapshot_id, snapshot_id);
+            assert!(reference.is_branch());
+        } else {
+            panic!("Expected SetSnapshotRef for main branch as second update");
+        }
+
+        // Third update: SetSnapshotRef for tag
+        if let TableUpdate::SetSnapshotRef {
+            ref_name,
+            reference,
+        } = &updates[2]
+        {
+            assert_eq!(ref_name, "v1.0.0");
+            assert_eq!(reference.snapshot_id, snapshot_id);
+            assert!(!reference.is_branch()); // Should be a tag, not a branch
+        } else {
+            panic!("Expected SetSnapshotRef for tag as third update");
+        }
+    }
 }

crates/iceberg/src/transaction/snapshot.rs

@@ -118,6 +118,8 @@ pub(crate) struct SnapshotProducer<'a> {
     // It starts from 0 and increments for each new manifest file.
     // Note: This counter is limited to the range of (0..u64::MAX).
     manifest_counter: RangeFrom<u64>,
+    // Optional tag name to create atomically with the snapshot.
+    tag_ref: Option<String>,
 }
 
 impl<'a> SnapshotProducer<'a> {
@@ -127,6 +129,7 @@ impl<'a> SnapshotProducer<'a> {
         key_metadata: Option<Vec<u8>>,
         snapshot_properties: HashMap<String, String>,
         added_data_files: Vec<DataFile>,
+        tag_ref: Option<String>,
     ) -> Self {
         Self {
             table,
@@ -136,6 +139,7 @@ impl<'a> SnapshotProducer<'a> {
             snapshot_properties,
             added_data_files,
             manifest_counter: (0..),
+            tag_ref,
         }
     }
 
@@ -485,7 +489,7 @@ impl<'a> SnapshotProducer<'a> {
             new_snapshot.build()
         };
 
-        let updates = vec![
+        let mut updates = vec![
             TableUpdate::AddSnapshot {
                 snapshot: new_snapshot,
             },
@@ -498,6 +502,15 @@ impl<'a> SnapshotProducer<'a> {
             },
         ];
 
+        if let Some(tag_name) = self.tag_ref {
+            updates.push(TableUpdate::SetSnapshotRef {
+                ref_name: tag_name,
+                reference: SnapshotReference::new(self.snapshot_id, SnapshotRetention::Tag {
+                    max_ref_age_ms: None,
+                }),
+            });
+        }
+
         let requirements = vec![
             TableRequirement::UuidMatch {
                 uuid: self.table.metadata().uuid(),