RemovePartitionSpecs update

crates/iceberg/src/catalog/mod.rs

@@ -470,6 +470,12 @@ pub enum TableUpdate {
         /// Snapshot id to remove partition statistics for.
         snapshot_id: i64,
     },
+    /// Remove partition specs
+    #[serde(rename_all = "kebab-case")]
+    RemovePartitionSpecs {
+        /// Partition spec ids to remove.
+        spec_ids: Vec<i32>,
+    },
 }
 
 impl TableUpdate {
@@ -510,6 +516,9 @@ impl TableUpdate {
             TableUpdate::RemovePartitionStatistics { snapshot_id } => {
                 Ok(builder.remove_partition_statistics(snapshot_id))
             }
+            TableUpdate::RemovePartitionSpecs { spec_ids } => {
+                Ok(builder.remove_partition_specs(&spec_ids))
+            }
         }
     }
 }
@@ -2019,4 +2028,17 @@ mod tests {
             },
         )
     }
+    fn test_remove_partition_specs_update() {
+        test_serde_json(
+            r#"
+{
+    "action": "remove-partition-specs",
+    "spec-ids": [1, 2]
+}        
+        "#,
+            TableUpdate::RemovePartitionSpecs {
+                spec_ids: vec![1, 2],
+            },
+        );
+    }
 }

crates/iceberg/src/spec/table_metadata_builder.rs

@@ -786,6 +786,32 @@ impl TableMetadataBuilder {
             .set_default_partition_spec(Self::LAST_ADDED)
     }
 
+    /// Remove partition specs by their ids from the table metadata.
+    /// Does nothing if a spec id is not present.
+    ///
+    /// If the default partition spec is removed, it is re-added
+    /// upon build.
+    pub fn remove_partition_specs(mut self, spec_ids: &[i32]) -> Self {
+        let mut removed_specs = Vec::with_capacity(spec_ids.len());
+
+        self.metadata.partition_specs.retain(|k, _| {
+            if spec_ids.contains(k) {
+                removed_specs.push(*k);
+                false
+            } else {
+                true
+            }
+        });
+
+        if !removed_specs.is_empty() {
+            self.changes.push(TableUpdate::RemovePartitionSpecs {
+                spec_ids: removed_specs,
+            });
+        }
+
+        self
+    }
+
     /// Add a sort order to the table metadata.
     ///
     /// The spec is bound eagerly to the current schema and must be valid for it.
@@ -1592,6 +1618,20 @@ mod tests {
         pretty_assertions::assert_eq!(build_result.changes[0], TableUpdate::AddSpec {
             spec: expected_change
         });
+
+        // Remove the spec
+        let build_result = build_result
+            .metadata
+            .into_builder(Some(
+                "s3://bucket/test/location/metadata/metadata1.json".to_string(),
+            ))
+            .remove_partition_specs(&[1])
+            .build()
+            .unwrap();
+
+        assert_eq!(build_result.changes.len(), 1);
+        assert_eq!(build_result.metadata.partition_specs.len(), 1);
+        assert!(build_result.metadata.partition_spec_by_id(1).is_none());
     }
 
     #[test]
@@ -2297,5 +2337,18 @@ mod tests {
             .unwrap();
         assert_eq!(build_result.metadata.partition_statistics.len(), 0);
         assert_eq!(build_result.changes.len(), 0);
+        fn test_default_spec_cannot_be_removed() {
+            let builder = builder_without_changes(FormatVersion::V2);
+
+            let metadata = builder
+                .remove_partition_specs(&[0])
+                .build()
+                .unwrap()
+                .metadata;
+
+            assert_eq!(metadata.default_spec.spec_id(), 0);
+            assert_eq!(metadata.partition_specs.len(), 1);
+            assert!(metadata.partition_spec_by_id(0).is_some());
+        }
     }
 }