WIP

Author: c-thiel

crates/catalog/glue/src/catalog.rs

@@ -355,7 +355,9 @@ impl Catalog for GlueCatalog {
             }
         };
 
-        let metadata = TableMetadataBuilder::from_table_creation(creation)?.build()?;
+        let metadata = TableMetadataBuilder::from_table_creation(creation)?
+            .build()?
+            .metadata;
         let metadata_location = create_metadata_location(&location, 0)?;
 
         self.file_io

crates/catalog/glue/src/schema.rs

@@ -198,7 +198,9 @@ mod tests {
             .location("my_location".to_string())
             .schema(schema)
             .build();
-        let metadata = TableMetadataBuilder::from_table_creation(table_creation)?.build()?;
+        let metadata = TableMetadataBuilder::from_table_creation(table_creation)?
+            .build()?
+            .metadata;
 
         Ok(metadata)
     }

crates/catalog/glue/src/utils.rs

@@ -299,7 +299,9 @@ mod tests {
             .location("my_location".to_string())
             .schema(schema)
             .build();
-        let metadata = TableMetadataBuilder::from_table_creation(table_creation)?.build()?;
+        let metadata = TableMetadataBuilder::from_table_creation(table_creation)?
+            .build()?
+            .metadata;
 
         Ok(metadata)
     }

crates/catalog/hms/src/catalog.rs

@@ -346,7 +346,9 @@ impl Catalog for HmsCatalog {
             }
         };
 
-        let metadata = TableMetadataBuilder::from_table_creation(creation)?.build()?;
+        let metadata = TableMetadataBuilder::from_table_creation(creation)?
+            .build()?
+            .metadata;
         let metadata_location = create_metadata_location(&location, 0)?;
 
         self.file_io

crates/catalog/memory/src/catalog.rs

@@ -194,7 +194,9 @@ impl Catalog for MemoryCatalog {
             }
         };
 
-        let metadata = TableMetadataBuilder::from_table_creation(table_creation)?.build()?;
+        let metadata = TableMetadataBuilder::from_table_creation(table_creation)?
+            .build()?
+            .metadata;
         let metadata_location = format!(
             "{}/metadata/{}-{}.metadata.json",
             &location,
@@ -371,7 +373,7 @@ mod tests {
         let expected_sorted_order = SortOrder::builder()
             .with_order_id(0)
             .with_fields(vec![])
-            .build(expected_schema.clone())
+            .build(&expected_schema)
             .unwrap();
 
         assert_eq!(

crates/iceberg/src/catalog/mod.rs

@@ -433,8 +433,46 @@ impl TableUpdate {
     /// Applies the update to the table metadata builder.
     pub fn apply(self, builder: TableMetadataBuilder) -> Result<TableMetadataBuilder> {
         match self {
-            TableUpdate::AssignUuid { uuid } => builder.assign_uuid(uuid),
-            _ => unimplemented!(),
+            TableUpdate::AssignUuid { uuid } => Ok(builder.assign_uuid(uuid)),
+            TableUpdate::AddSchema {
+                schema,
+                last_column_id,
+            } => {
+                if let Some(last_column_id) = last_column_id {
+                    if builder.last_column_id() < last_column_id {
+                        return Err(Error::new(
+                            ErrorKind::DataInvalid,
+                            format!(
+                                "Invalid last column ID: {last_column_id} < {} (previous last column ID)",
+                                builder.last_column_id()
+                            ),
+                        ));
+                    }
+                };
+                builder.add_schema(schema)
+            }
+            TableUpdate::SetCurrentSchema { schema_id } => builder.set_current_schema(schema_id),
+            TableUpdate::AddSpec { spec } => builder.add_partition_spec(spec),
+            TableUpdate::SetDefaultSpec { spec_id } => builder.set_default_partition_spec(spec_id),
+            TableUpdate::AddSortOrder { sort_order } => builder.add_sort_order(sort_order),
+            TableUpdate::SetDefaultSortOrder { sort_order_id } => {
+                builder.set_default_sort_order(sort_order_id)
+            }
+            TableUpdate::AddSnapshot { snapshot } => builder.add_snapshot(snapshot),
+            TableUpdate::SetSnapshotRef {
+                ref_name,
+                reference,
+            } => builder.set_ref(&ref_name, reference),
+            TableUpdate::RemoveSnapshots { snapshot_ids } => {
+                Ok(builder.remove_snapshots(&snapshot_ids))
+            }
+            TableUpdate::RemoveSnapshotRef { ref_name } => Ok(builder.remove_ref(&ref_name)),
+            TableUpdate::SetLocation { location } => Ok(builder.set_location(location)),
+            TableUpdate::SetProperties { updates } => builder.set_properties(updates),
+            TableUpdate::RemoveProperties { removals } => Ok(builder.remove_properties(&removals)),
+            TableUpdate::UpgradeFormatVersion { format_version } => {
+                builder.upgrade_format_version(format_version)
+            }
         }
     }
 }
@@ -1102,16 +1140,21 @@ mod tests {
         let table_metadata = TableMetadataBuilder::from_table_creation(table_creation)
             .unwrap()
             .build()
-            .unwrap();
-        let table_metadata_builder = TableMetadataBuilder::new(table_metadata);
+            .unwrap()
+            .metadata;
+        let table_metadata_builder = TableMetadataBuilder::new_from_metadata(
+            table_metadata,
+            "s3://db/table/metadata/metadata1.gz.json",
+        );
 
         let uuid = uuid::Uuid::new_v4();
         let update = TableUpdate::AssignUuid { uuid };
         let updated_metadata = update
             .apply(table_metadata_builder)
             .unwrap()
             .build()
-            .unwrap();
+            .unwrap()
+            .metadata;
         assert_eq!(updated_metadata.uuid(), uuid);
     }
 }

crates/iceberg/src/spec/partition.rs

@@ -150,7 +150,7 @@ impl PartitionSpec {
 
     /// 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
+    /// This is required for spec version 1
     pub fn has_sequential_ids(&self) -> bool {
         for (index, field) in self.fields.iter().enumerate() {
             let expected_id = (UNPARTITIONED_LAST_ASSIGNED_ID as i64)

crates/iceberg/src/spec/schema.rs

@@ -39,7 +39,7 @@ use crate::{ensure_data_valid, Error, ErrorKind};
 pub type SchemaId = i32;
 /// Reference to [`Schema`].
 pub type SchemaRef = Arc<Schema>;
-const DEFAULT_SCHEMA_ID: SchemaId = 0;
+pub(crate) const DEFAULT_SCHEMA_ID: SchemaId = 0;
 
 /// Defines schema in iceberg.
 #[derive(Debug, Serialize, Deserialize, Clone)]
@@ -133,7 +133,7 @@ impl SchemaBuilder {
         let r#struct = StructType::new(self.fields);
         let id_to_field = index_by_id(&r#struct)?;
         let highest_field_id =
-            highest_field_id.unwrap_or(get_highest_schema_id_for_struct(&r#struct));
+            highest_field_id.unwrap_or(id_to_field.keys().max().cloned().unwrap_or(0));
 
         Self::validate_identifier_ids(
             &r#struct,
@@ -367,13 +367,18 @@ impl Schema {
         self.id_to_name.get(&field_id).map(String::as_str)
     }
 
+    /// Return A HashMap matching field ids to field names.
+    pub(crate) fn field_id_to_name_map(&self) -> &HashMap<i32, String> {
+        &self.id_to_name
+    }
+
     /// Get an accessor for retrieving data in a struct
     pub fn accessor_by_field_id(&self, field_id: i32) -> Option<Arc<StructAccessor>> {
         self.field_id_to_accessor.get(&field_id).cloned()
     }
 
     /// Check if this schema is identical to another schema semantically - excluding schema id.
-    pub fn is_same_schema(&self, other: &SchemaRef) -> bool {
+    pub(crate) fn is_same_schema(&self, other: &SchemaRef) -> bool {
         self.as_struct().eq(other.as_struct())
             && self.identifier_field_ids().eq(other.identifier_field_ids())
     }
@@ -981,58 +986,6 @@ impl SchemaVisitor for PruneColumn {
     }
 }
 
-/// Get the highest field id in a struct.
-/// Recursively visits nested fields.
-pub fn get_highest_schema_id_for_struct(s: &StructType) -> i32 {
-    let mut agg = HighestFieldIdAggregator::new();
-    visit_struct(s, &mut agg).unwrap_or_default();
-    agg.highest_field_id
-}
-
-struct HighestFieldIdAggregator {
-    highest_field_id: i32,
-}
-
-impl HighestFieldIdAggregator {
-    fn new() -> Self {
-        Self {
-            highest_field_id: 0,
-        }
-    }
-}
-
-impl SchemaVisitor for HighestFieldIdAggregator {
-    type T = ();
-
-    fn schema(&mut self, _schema: &Schema, _value: Self::T) -> Result<Self::T> {
-        Ok(())
-    }
-
-    fn field(&mut self, field: &NestedFieldRef, _value: Self::T) -> Result<Self::T> {
-        self.highest_field_id = std::cmp::max(self.highest_field_id, field.id);
-        Ok(())
-    }
-
-    fn r#struct(&mut self, _struct: &StructType, _results: Vec<Self::T>) -> Result<Self::T> {
-        Ok(())
-    }
-
-    fn list(&mut self, list: &ListType, _value: Self::T) -> Result<Self::T> {
-        self.highest_field_id = std::cmp::max(self.highest_field_id, list.element_field.id);
-        Ok(())
-    }
-
-    fn map(&mut self, map: &MapType, _key_value: Self::T, _value: Self::T) -> Result<Self::T> {
-        self.highest_field_id = std::cmp::max(self.highest_field_id, map.key_field.id);
-        self.highest_field_id = std::cmp::max(self.highest_field_id, map.value_field.id);
-        Ok(())
-    }
-
-    fn primitive(&mut self, _p: &PrimitiveType) -> Result<Self::T> {
-        Ok(())
-    }
-}
-
 struct ReassignFieldIds {
     next_field_id: i32,
     old_to_new_id: HashMap<i32, i32>,

crates/iceberg/src/spec/snapshot.rs

@@ -22,11 +22,10 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use _serde::SnapshotV2;
-use chrono::{DateTime, TimeZone, Utc};
+use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use typed_builder::TypedBuilder;
 
-use super::table_metadata::SnapshotLog;
 use crate::error::{timestamp_ms_to_utc, Result};
 use crate::io::FileIO;
 use crate::spec::{ManifestList, SchemaId, SchemaRef, StructType, TableMetadata};
@@ -192,13 +191,6 @@ impl Snapshot {
             partition_type_provider,
         )
     }
-
-    pub(crate) fn log(&self) -> SnapshotLog {
-        SnapshotLog {
-            timestamp_ms: self.timestamp_ms,
-            snapshot_id: self.snapshot_id,
-        }
-    }
 }
 
 pub(super) mod _serde {

crates/iceberg/src/spec/sort.rs

@@ -138,7 +138,7 @@ impl SortOrder {
 impl SortOrderBuilder {
     /// Creates a new unbound sort order.
     pub fn build_unbound(&self) -> Result<SortOrder> {
-        let fields = self.fields.clone().unwrap_or_default();
+        let fields: Vec<SortField> = self.fields.clone().unwrap_or_default();
         return match (self.order_id, fields.as_slice()) {
             (Some(SortOrder::UNSORTED_ORDER_ID) | None, []) => Ok(SortOrder::unsorted_order()),
             (_, []) => Err(Error::new(
@@ -160,13 +160,13 @@ impl SortOrderBuilder {
     }
 
     /// Creates a new bound sort order.
-    pub fn build(&self, schema: Schema) -> Result<SortOrder> {
+    pub fn build(&self, schema: &Schema) -> Result<SortOrder> {
         let unbound_sort_order = self.build_unbound()?;
         SortOrderBuilder::check_compatibility(unbound_sort_order, schema)
     }
 
     /// Returns the given sort order if it is compatible with the given schema
-    fn check_compatibility(sort_order: SortOrder, schema: Schema) -> Result<SortOrder> {
+    fn check_compatibility(sort_order: SortOrder, schema: &Schema) -> Result<SortOrder> {
         let sort_fields = &sort_order.fields;
         for sort_field in sort_fields {
             match schema.field_by_id(sort_field.source_id) {
@@ -290,6 +290,35 @@ mod tests {
         )
     }
 
+    #[test]
+    fn test_build_unbound_returns_correct_default_order_id_for_no_fields() {
+        assert_eq!(
+            SortOrder::builder()
+                .build_unbound()
+                .expect("Expected an Ok value")
+                .order_id,
+            SortOrder::UNSORTED_ORDER_ID
+        )
+    }
+
+    #[test]
+    fn test_build_unbound_returns_correct_default_order_id_for_fields() {
+        let sort_field = SortField::builder()
+            .source_id(2)
+            .direction(SortDirection::Ascending)
+            .null_order(NullOrder::First)
+            .transform(Transform::Identity)
+            .build();
+        assert_ne!(
+            SortOrder::builder()
+                .with_sort_field(sort_field.clone())
+                .build_unbound()
+                .expect("Expected an Ok value")
+                .order_id,
+            SortOrder::UNSORTED_ORDER_ID
+        )
+    }
+
     #[test]
     fn test_build_unbound_should_return_unsorted_sort_order() {
         assert_eq!(
@@ -367,7 +396,7 @@ mod tests {
                     .transform(Transform::Identity)
                     .build(),
             )
-            .build(schema);
+            .build(&schema);
 
         assert_eq!(
             sort_order_builder_result
@@ -406,7 +435,7 @@ mod tests {
                     .transform(Transform::Identity)
                     .build(),
             )
-            .build(schema);
+            .build(&schema);
 
         assert_eq!(
             sort_order_builder_result
@@ -438,7 +467,7 @@ mod tests {
                     .transform(Transform::Year)
                     .build(),
             )
-            .build(schema);
+            .build(&schema);
 
         assert_eq!(
             sort_order_builder_result
@@ -468,7 +497,7 @@ mod tests {
 
         let sort_order_builder_result = SortOrder::builder()
             .with_sort_field(sort_field.clone())
-            .build(schema);
+            .build(&schema);
 
         assert_eq!(
             sort_order_builder_result.expect("Expected an Ok value"),