feat(transaction): add ManageSnapshotsAction for snapshot refs. (#1939)

crates/iceberg/public-api.txt

@@ -2617,10 +2617,12 @@ pub fn iceberg::spec::TableMetadata::partition_statistics_for_snapshot(&self, sn
 pub fn iceberg::spec::TableMetadata::partition_statistics_iter(&self) -> impl core::iter::traits::exact_size::ExactSizeIterator<Item = &iceberg::spec::PartitionStatisticsFile>
 pub fn iceberg::spec::TableMetadata::properties(&self) -> &std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>
 pub async fn iceberg::spec::TableMetadata::read_from(file_io: &iceberg::io::FileIO, metadata_location: impl core::convert::AsRef<str>) -> iceberg::Result<iceberg::spec::TableMetadata>
+pub fn iceberg::spec::TableMetadata::refs(&self) -> &std::collections::hash::map::HashMap<alloc::string::String, iceberg::spec::SnapshotReference>
 pub fn iceberg::spec::TableMetadata::schema_by_id(&self, schema_id: iceberg::spec::SchemaId) -> core::option::Option<&iceberg::spec::SchemaRef>
 pub fn iceberg::spec::TableMetadata::schemas_iter(&self) -> impl core::iter::traits::exact_size::ExactSizeIterator<Item = &iceberg::spec::SchemaRef>
 pub fn iceberg::spec::TableMetadata::snapshot_by_id(&self, snapshot_id: i64) -> core::option::Option<&iceberg::spec::SnapshotRef>
 pub fn iceberg::spec::TableMetadata::snapshot_for_ref(&self, ref_name: &str) -> core::option::Option<&iceberg::spec::SnapshotRef>
+pub fn iceberg::spec::TableMetadata::snapshot_reference(&self, name: &str) -> core::option::Option<&iceberg::spec::SnapshotReference>
 pub fn iceberg::spec::TableMetadata::snapshots(&self) -> impl core::iter::traits::exact_size::ExactSizeIterator<Item = &iceberg::spec::SnapshotRef>
 pub fn iceberg::spec::TableMetadata::sort_order_by_id(&self, sort_order_id: i64) -> core::option::Option<&iceberg::spec::SortOrderRef>
 pub fn iceberg::spec::TableMetadata::sort_orders_iter(&self) -> impl core::iter::traits::exact_size::ExactSizeIterator<Item = &iceberg::spec::SortOrderRef>
@@ -3080,6 +3082,7 @@ impl iceberg::transaction::Transaction
 pub async fn iceberg::transaction::Transaction::commit(self, catalog: &dyn iceberg::Catalog) -> iceberg::Result<iceberg::table::Table>
 pub fn iceberg::transaction::Transaction::expire_snapshots(&self) -> iceberg::transaction::expire_snapshots::ExpireSnapshotsAction
 pub fn iceberg::transaction::Transaction::fast_append(&self) -> iceberg::transaction::append::FastAppendAction
+pub fn iceberg::transaction::Transaction::manage_snapshots(&self) -> iceberg::transaction::manage_snapshots::ManageSnapshotsAction
 pub fn iceberg::transaction::Transaction::new(table: &iceberg::table::Table) -> Self
 pub fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::sort_order::ReplaceSortOrderAction
 pub fn iceberg::transaction::Transaction::update_location(&self) -> iceberg::transaction::update_location::UpdateLocationAction
@@ -3104,6 +3107,7 @@ pub mod iceberg::util
 pub mod iceberg::util::snapshot
 pub fn iceberg::util::snapshot::ancestors_between(table_metadata: &iceberg::spec::TableMetadataRef, latest_snapshot_id: i64, oldest_snapshot_id: core::option::Option<i64>) -> impl core::iter::traits::iterator::Iterator<Item = iceberg::spec::SnapshotRef> + core::marker::Send
 pub fn iceberg::util::snapshot::ancestors_of(table_metadata: &iceberg::spec::TableMetadataRef, snapshot_id: i64) -> impl core::iter::traits::iterator::Iterator<Item = iceberg::spec::SnapshotRef> + core::marker::Send
+pub fn iceberg::util::snapshot::is_ancestor_of(table_metadata: &iceberg::spec::TableMetadataRef, descendant_id: i64, ancestor_id: i64) -> bool
 pub mod iceberg::writer
 pub mod iceberg::writer::base_writer
 pub mod iceberg::writer::base_writer::data_file_writer

crates/iceberg/src/spec/table_metadata.rs

@@ -332,6 +332,20 @@ impl TableMetadata {
         })
     }
 
+    /// All snapshot references (branches and tags) keyed by name.
+    #[inline]
+    pub fn refs(&self) -> &HashMap<String, SnapshotReference> {
+        &self.refs
+    }
+
+    /// The snapshot reference (snapshot id and retention policy) for `name`, or `None` if there is
+    /// no such ref. Distinct from [`snapshot_for_ref`](Self::snapshot_for_ref), which resolves the
+    /// reference to its [`Snapshot`](crate::spec::Snapshot) and drops the retention metadata.
+    #[inline]
+    pub fn snapshot_reference(&self, name: &str) -> Option<&SnapshotReference> {
+        self.refs.get(name)
+    }
+
     /// Return all sort orders.
     #[inline]
     pub fn sort_orders_iter(&self) -> impl ExactSizeIterator<Item = &SortOrderRef> {

crates/iceberg/src/spec/table_metadata_builder.rs

@@ -534,6 +534,14 @@ impl TableMetadataBuilder {
             ));
         };
 
+        // The main branch must always be a branch, never a tag (Java `TableMetadata.Builder.setRef`).
+        if ref_name == MAIN_BRANCH && !reference.is_branch() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!("Cannot set {MAIN_BRANCH} to a tag, it must be a branch"),
+            ));
+        }
+
         // Update last_updated_ms to the exact timestamp of the snapshot if it was added in this commit
         let is_added_snapshot = self.changes.iter().any(|update| {
             matches!(update, TableUpdate::AddSnapshot { snapshot: snap } if snap.snapshot_id() == snapshot.snapshot_id())
@@ -2212,6 +2220,40 @@ mod tests {
         }])
     }
 
+    #[test]
+    fn test_set_main_ref_to_tag_fails() {
+        let builder = builder_without_changes(FormatVersion::V2);
+
+        let snapshot = Snapshot::builder()
+            .with_snapshot_id(1)
+            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
+            .with_sequence_number(0)
+            .with_schema_id(0)
+            .with_manifest_list("/snap-1.avro")
+            .with_summary(Summary {
+                operation: Operation::Append,
+                additional_properties: HashMap::new(),
+            })
+            .build();
+
+        // The main branch must be a branch; setting it to a tag is rejected (Java parity).
+        let err = builder
+            .add_snapshot(snapshot)
+            .unwrap()
+            .set_ref(MAIN_BRANCH, SnapshotReference {
+                snapshot_id: 1,
+                retention: SnapshotRetention::Tag {
+                    max_ref_age_ms: None,
+                },
+            })
+            .unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("Cannot set main to a tag, it must be a branch"),
+            "unexpected error: {err}"
+        );
+    }
+
     #[test]
     fn test_snapshot_log_skips_intermediates() {
         let builder = builder_without_changes(FormatVersion::V2);