WIP - alot with schemas

Author: c-thiel

crates/catalog/rest/tests/rest_catalog_test.rs

@@ -293,12 +293,8 @@ async fn test_create_table() {
     assert_eq!(table.metadata().format_version(), FormatVersion::V2);
     assert!(table.metadata().current_snapshot().is_none());
     assert!(table.metadata().history().is_empty());
-    assert!(table.metadata().default_sort_order().unwrap().is_unsorted());
-    assert!(table
-        .metadata()
-        .default_partition_spec()
-        .unwrap()
-        .is_unpartitioned());
+    assert!(table.metadata().default_sort_order().is_unsorted());
+    assert!(table.metadata().default_partition_spec().is_unpartitioned());
 }
 
 #[tokio::test]

crates/iceberg/src/error.rs

@@ -47,7 +47,7 @@ pub enum ErrorKind {
     /// Validation failed.
     ///
     /// This error is returned when Table or View Metadata is manipulated
-    /// in non-allowed ways.
+    /// in forbidden ways that would produce corrupt metadata.
     ValidationFailed,
 }
 

crates/iceberg/src/io/object_cache.rs

@@ -252,7 +252,7 @@ mod tests {
         async fn setup_manifest_files(&mut self) {
             let current_snapshot = self.table.metadata().current_snapshot().unwrap();
             let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
-            let current_partition_spec = self.table.metadata().default_partition_spec().unwrap();
+            let current_partition_spec = self.table.metadata().default_partition_spec();
 
             // Write data files
             let data_file_manifest = ManifestWriter::new(

crates/iceberg/src/scan.rs

@@ -966,7 +966,7 @@ mod tests {
                 .parent_snapshot(self.table.metadata())
                 .unwrap();
             let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
-            let current_partition_spec = self.table.metadata().default_partition_spec().unwrap();
+            let current_partition_spec = self.table.metadata().default_partition_spec();
 
             // Write data files
             let data_file_manifest = ManifestWriter::new(

crates/iceberg/src/spec/datatypes.rs

@@ -668,6 +668,12 @@ impl NestedField {
         self.write_default = Some(value);
         self
     }
+
+    /// Set the id of the field.
+    pub(crate) fn with_id(mut self, id: i32) -> Self {
+        self.id = id;
+        self
+    }
 }
 
 impl fmt::Display for NestedField {

crates/iceberg/src/spec/partition.rs

@@ -121,6 +121,60 @@ impl PartitionSpec {
     pub fn to_unbound(self) -> UnboundPartitionSpec {
         self.into()
     }
+
+    /// Check if this partition spec is compatible with another partition spec.
+    ///
+    /// Returns true if the partition spec is equal to the other spec with partition field ids ignored and
+    /// spec_id ignored. The following must be identical:
+    /// * The number of fields
+    /// * Field order
+    /// * Field names
+    /// * Source column ids
+    /// * Transforms
+    pub fn compatible_with(&self, other: &UnboundPartitionSpec) -> bool {
+        if self.fields.len() != other.fields.len() {
+            return false;
+        }
+
+        for (this_field, other_field) in self.fields.iter().zip(&other.fields) {
+            if this_field.source_id != other_field.source_id
+                || this_field.transform != other_field.transform
+                || this_field.name != other_field.name
+            {
+                return false;
+            }
+        }
+
+        true
+    }
+
+    /// Check if this partition spec has sequential partition ids.
+    /// Sequential ids start from 1000 and increment by 1 for each field.
+    /// This is required for spec version
+    pub fn has_sequential_ids(&self) -> bool {
+        for (index, field) in self.fields.iter().enumerate() {
+            let expected_id = (UNPARTITIONED_LAST_ASSIGNED_ID as i64)
+                .checked_add(1)
+                .and_then(|id| id.checked_add(index as i64))
+                .unwrap_or(i64::MAX);
+
+            if field.field_id as i64 != expected_id {
+                return false;
+            }
+        }
+
+        true
+    }
+
+    /// Get the highest field id in the partition spec.
+    /// If the partition spec is unpartitioned, it returns the last unpartitioned last assigned id (999).
+    pub fn highest_field_id(&self) -> i32 {
+        self.fields
+            .iter()
+            .map(|f| f.field_id)
+            .max()
+            .unwrap_or(UNPARTITIONED_LAST_ASSIGNED_ID)
+    }
 }
 
 /// Reference to [`UnboundPartitionSpec`].
@@ -1263,4 +1317,7 @@ mod tests {
             }]
         });
     }
+
+    #[test]
+    fn test_has_sequential_ids() {}
 }