fix: Use CatalogCommitConflicts for TableRequirement (#1532)

## Which issue does this PR close?

- Closes #1531 

## What changes are included in this PR?


- Use `ErrorKind::CatalogCommitConflicts` for `TableRequirement::check`
- When metadata is None, returns `ErrorKind::TableNotFound`
- Minor trimmings

## Are these changes tested?
No

Author: CTTY

crates/iceberg/src/catalog/mod.rs

@@ -597,111 +597,120 @@ impl TableRequirement {
             match self {
                 TableRequirement::NotExist => {
                     return Err(Error::new(
-                        ErrorKind::DataInvalid,
+                        ErrorKind::CatalogCommitConflicts,
                         format!(
                             "Requirement failed: Table with id {} already exists",
                             metadata.uuid()
                         ),
-                    ));
+                    )
+                    .with_retryable(true));
                 }
                 TableRequirement::UuidMatch { uuid } => {
                     if &metadata.uuid() != uuid {
                         return Err(Error::new(
-                            ErrorKind::DataInvalid,
+                            ErrorKind::CatalogCommitConflicts,
                             "Requirement failed: Table UUID does not match",
                         )
                         .with_context("expected", *uuid)
-                        .with_context("found", metadata.uuid()));
+                        .with_context("found", metadata.uuid())
+                        .with_retryable(true));
                     }
                 }
                 TableRequirement::CurrentSchemaIdMatch { current_schema_id } => {
                     // ToDo: Harmonize the types of current_schema_id
                     if metadata.current_schema_id != *current_schema_id {
                         return Err(Error::new(
-                            ErrorKind::DataInvalid,
+                            ErrorKind::CatalogCommitConflicts,
                             "Requirement failed: Current schema id does not match",
                         )
                         .with_context("expected", current_schema_id.to_string())
-                        .with_context("found", metadata.current_schema_id.to_string()));
+                        .with_context("found", metadata.current_schema_id.to_string())
+                        .with_retryable(true));
                     }
                 }
                 TableRequirement::DefaultSortOrderIdMatch {
                     default_sort_order_id,
                 } => {
                     if metadata.default_sort_order().order_id != *default_sort_order_id {
                         return Err(Error::new(
-                            ErrorKind::DataInvalid,
+                            ErrorKind::CatalogCommitConflicts,
                             "Requirement failed: Default sort order id does not match",
                         )
                         .with_context("expected", default_sort_order_id.to_string())
-                        .with_context(
-                            "found",
-                            metadata.default_sort_order().order_id.to_string(),
-                        ));
+                        .with_context("found", metadata.default_sort_order().order_id.to_string())
+                        .with_retryable(true));
                     }
                 }
                 TableRequirement::RefSnapshotIdMatch { r#ref, snapshot_id } => {
                     let snapshot_ref = metadata.snapshot_for_ref(r#ref);
                     if let Some(snapshot_id) = snapshot_id {
-                        let snapshot_ref = snapshot_ref.ok_or(Error::new(
-                            ErrorKind::DataInvalid,
-                            format!("Requirement failed: Branch or tag `{}` not found", r#ref),
-                        ))?;
+                        let snapshot_ref = snapshot_ref.ok_or(
+                            Error::new(
+                                ErrorKind::CatalogCommitConflicts,
+                                format!("Requirement failed: Branch or tag `{}` not found", r#ref),
+                            )
+                            .with_retryable(true),
+                        )?;
                         if snapshot_ref.snapshot_id() != *snapshot_id {
                             return Err(Error::new(
-                                ErrorKind::DataInvalid,
+                                ErrorKind::CatalogCommitConflicts,
                                 format!(
                                     "Requirement failed: Branch or tag `{}`'s snapshot has changed",
                                     r#ref
                                 ),
                             )
                             .with_context("expected", snapshot_id.to_string())
-                            .with_context("found", snapshot_ref.snapshot_id().to_string()));
+                            .with_context("found", snapshot_ref.snapshot_id().to_string())
+                            .with_retryable(true));
                         }
                     } else if snapshot_ref.is_some() {
                         // a null snapshot ID means the ref should not exist already
                         return Err(Error::new(
-                            ErrorKind::DataInvalid,
+                            ErrorKind::CatalogCommitConflicts,
                             format!(
                                 "Requirement failed: Branch or tag `{}` already exists",
                                 r#ref
                             ),
-                        ));
+                        )
+                        .with_retryable(true));
                     }
                 }
                 TableRequirement::DefaultSpecIdMatch { default_spec_id } => {
                     // ToDo: Harmonize the types of default_spec_id
                     if metadata.default_partition_spec_id() != *default_spec_id {
                         return Err(Error::new(
-                            ErrorKind::DataInvalid,
+                            ErrorKind::CatalogCommitConflicts,
                             "Requirement failed: Default partition spec id does not match",
                         )
                         .with_context("expected", default_spec_id.to_string())
-                        .with_context("found", metadata.default_partition_spec_id().to_string()));
+                        .with_context("found", metadata.default_partition_spec_id().to_string())
+                        .with_retryable(true));
                     }
                 }
                 TableRequirement::LastAssignedPartitionIdMatch {
                     last_assigned_partition_id,
                 } => {
                     if metadata.last_partition_id != *last_assigned_partition_id {
                         return Err(Error::new(
-                            ErrorKind::DataInvalid,
+                            ErrorKind::CatalogCommitConflicts,
                             "Requirement failed: Last assigned partition id does not match",
                         )
                         .with_context("expected", last_assigned_partition_id.to_string())
-                        .with_context("found", metadata.last_partition_id.to_string()));
+                        .with_context("found", metadata.last_partition_id.to_string())
+                        .with_retryable(true));
                     }
                 }
                 TableRequirement::LastAssignedFieldIdMatch {
                     last_assigned_field_id,
                 } => {
                     if &metadata.last_column_id != last_assigned_field_id {
                         return Err(Error::new(
-                            ErrorKind::DataInvalid,
+                            ErrorKind::CatalogCommitConflicts,
                             "Requirement failed: Last assigned field id does not match",
                         )
                         .with_context("expected", last_assigned_field_id.to_string())
-                        .with_context("found", metadata.last_column_id.to_string()));
+                        .with_context("found", metadata.last_column_id.to_string())
+                        .with_retryable(true));
                     }
                 }
             };
@@ -710,7 +719,7 @@ impl TableRequirement {
                 TableRequirement::NotExist => {}
                 _ => {
                     return Err(Error::new(
-                        ErrorKind::DataInvalid,
+                        ErrorKind::TableNotFound,
                         "Requirement failed: Table does not exist",
                     ));
                 }
@@ -814,7 +823,7 @@ pub enum ViewUpdate {
     #[serde(rename_all = "kebab-case")]
     AssignUuid {
         /// The new UUID to assign.
-        uuid: uuid::Uuid,
+        uuid: Uuid,
     },
     /// Upgrade view's format version
     #[serde(rename_all = "kebab-case")]
@@ -1092,7 +1101,7 @@ mod tests {
         .unwrap()
         .metadata;
 
-        // Ref exists and should matches
+        // Ref exists and should match
         let requirement = TableRequirement::RefSnapshotIdMatch {
             r#ref: "main".to_string(),
             snapshot_id: Some(3051729675574597004),

crates/iceberg/src/error.rs

@@ -53,7 +53,7 @@ pub enum ErrorKind {
     /// Iceberg namespace already exists at creation.
     NamespaceNotFound,
 
-    /// Iceberg table already exists at creation.
+    /// Iceberg table does not exist.
     TableNotFound,
 
     /// Iceberg feature is not supported.
@@ -278,7 +278,7 @@ impl Error {
     /// If you just want to print error with backtrace, use `Debug`, like `format!("{err:?}")`.
     ///
     /// If you use nightly rust, and want to access `iceberg::Error`'s backtrace in the standard way, you can
-    /// implement a newtype like this:
+    /// implement a new type like this:
     ///
     /// ```ignore
     /// // assume you already have `#![feature(error_generic_member_access)]` on the top of your crate