feat(iceberg): introduce remove schemas (#1115)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

## What changes are included in this PR?


This PR introduces the capability to remove schemas to enable expire
snapshot action operations.


https://github.com/apache/iceberg/blob/456bbe98b0b0982278a61af4c44d32e1c27417e2/core/src/main/java/org/apache/iceberg/TableMetadata.java#L1177


https://github.com/apache/iceberg/blob/456bbe98b0b0982278a61af4c44d32e1c27417e2/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java#L249

refer to risingwavelabs#20


<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->

## Are these changes tested?

- UT

<!--
Specify what test covers (unit test, integration test, etc.).

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

Author: Li0k

crates/iceberg/src/catalog/mod.rs

@@ -483,6 +483,12 @@ pub enum TableUpdate {
         /// Snapshot id to remove partition statistics for.
         snapshot_id: i64,
     },
+    /// Remove schemas
+    #[serde(rename_all = "kebab-case")]
+    RemoveSchemas {
+        /// Schema IDs to remove.
+        schema_ids: Vec<i32>,
+    },
 }
 
 impl TableUpdate {
@@ -526,6 +532,7 @@ impl TableUpdate {
             TableUpdate::RemovePartitionStatistics { snapshot_id } => {
                 Ok(builder.remove_partition_statistics(snapshot_id))
             }
+            TableUpdate::RemoveSchemas { schema_ids } => builder.remove_schemas(&schema_ids),
         }
     }
 }
@@ -2049,4 +2056,19 @@ mod tests {
             },
         )
     }
+
+    #[test]
+    fn test_remove_schema_update() {
+        test_serde_json(
+            r#"
+                {
+                    "action": "remove-schemas",
+                    "schema-ids": [1, 2]
+                }        
+            "#,
+            TableUpdate::RemoveSchemas {
+                schema_ids: vec![1, 2],
+            },
+        );
+    }
 }

crates/iceberg/src/spec/table_metadata_builder.rs

@@ -1210,6 +1210,37 @@ impl TableMetadataBuilder {
     fn highest_sort_order_id(&self) -> Option<i64> {
         self.metadata.sort_orders.keys().max().copied()
     }
+
+    /// Remove schemas by their ids from the table metadata.
+    /// Does nothing if a schema id is not present. Active schemas should not be removed.
+    pub fn remove_schemas(mut self, schema_id_to_remove: &[i32]) -> Result<Self> {
+        if schema_id_to_remove.contains(&self.metadata.current_schema_id) {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "Cannot remove current schema",
+            ));
+        }
+
+        if schema_id_to_remove.is_empty() {
+            return Ok(self);
+        }
+
+        let mut removed_schemas = Vec::with_capacity(schema_id_to_remove.len());
+        self.metadata.schemas.retain(|id, _schema| {
+            if schema_id_to_remove.contains(id) {
+                removed_schemas.push(*id);
+                false
+            } else {
+                true
+            }
+        });
+
+        self.changes.push(TableUpdate::RemoveSchemas {
+            schema_ids: removed_schemas,
+        });
+
+        Ok(self)
+    }
 }
 
 impl From<TableMetadataBuildResult> for TableMetadata {
@@ -2412,4 +2443,54 @@ mod tests {
             table.metadata().current_snapshot_id().unwrap()
         );
     }
+
+    #[test]
+    fn test_active_schema_cannot_be_removed() {
+        let builder = builder_without_changes(FormatVersion::V2);
+        builder.remove_schemas(&[0]).unwrap_err();
+    }
+
+    #[test]
+    fn test_remove_schemas() {
+        let file = File::open(format!(
+            "{}/testdata/table_metadata/{}",
+            env!("CARGO_MANIFEST_DIR"),
+            "TableMetadataV2Valid.json"
+        ))
+        .unwrap();
+        let reader = BufReader::new(file);
+        let resp = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
+
+        let table = Table::builder()
+            .metadata(resp)
+            .metadata_location("s3://bucket/test/location/metadata/v1.json".to_string())
+            .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
+            .file_io(FileIOBuilder::new("memory").build().unwrap())
+            .build()
+            .unwrap();
+
+        assert_eq!(2, table.metadata().schemas.len());
+
+        {
+            // can not remove active schema
+            let meta_data_builder = table.metadata().clone().into_builder(None);
+            meta_data_builder.remove_schemas(&[1]).unwrap_err();
+        }
+
+        let mut meta_data_builder = table.metadata().clone().into_builder(None);
+        meta_data_builder = meta_data_builder.remove_schemas(&[0]).unwrap();
+        let build_result = meta_data_builder.build().unwrap();
+        assert_eq!(1, build_result.metadata.schemas.len());
+        assert_eq!(1, build_result.metadata.current_schema_id);
+        assert_eq!(1, build_result.metadata.current_schema().schema_id());
+        assert_eq!(1, build_result.changes.len());
+
+        let remove_schema_ids =
+            if let TableUpdate::RemoveSchemas { schema_ids } = &build_result.changes[0] {
+                schema_ids
+            } else {
+                unreachable!("Expected RemoveSchema change")
+            };
+        assert_eq!(remove_schema_ids, &[0]);
+    }
 }