Merge remote-tracking branch 'upstream/main'

Author: gruuya

iceberg-rust/src/catalog/commit.rs

@@ -9,8 +9,6 @@
 //! All changes are made atomically - either all updates succeed or none are applied.
 //! Requirements are checked first to ensure concurrent modifications don't corrupt state.
 
-use std::collections::HashMap;
-
 use iceberg_rust_spec::{
     spec::{
         partition::PartitionSpec,
@@ -24,6 +22,8 @@ use iceberg_rust_spec::{
     view_metadata::Materialization,
 };
 use serde_derive::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::time::{SystemTime, UNIX_EPOCH};
 use uuid::Uuid;
 
 use crate::error::Error;
@@ -417,6 +417,9 @@ pub fn apply_table_updates(
     metadata: &mut TableMetadata,
     updates: Vec<TableUpdate>,
 ) -> Result<(), Error> {
+    let mut added_schema_id = None;
+    let mut added_spec_id = None;
+    let mut added_sort_order_id = None;
     for update in updates {
         match update {
             TableUpdate::UpgradeFormatVersion { format_version } => {
@@ -431,25 +434,63 @@ pub fn apply_table_updates(
                 schema,
                 last_column_id,
             } => {
-                metadata.schemas.insert(*schema.schema_id(), schema);
+                let schema_id = *schema.schema_id();
+                metadata.schemas.insert(schema_id, schema);
+                added_schema_id = Some(schema_id);
                 if let Some(last_column_id) = last_column_id {
                     metadata.last_column_id = last_column_id;
                 }
             }
             TableUpdate::SetCurrentSchema { schema_id } => {
-                metadata.current_schema_id = schema_id;
+                if schema_id == -1 {
+                    if let Some(added_schema_id) = added_schema_id {
+                        metadata.current_schema_id = added_schema_id;
+                    } else {
+                        return Err(Error::InvalidFormat(
+                            "Cannot set current schema to -1 without adding a schema first"
+                                .to_string(),
+                        ));
+                    }
+                } else {
+                    metadata.current_schema_id = schema_id;
+                }
             }
             TableUpdate::AddSpec { spec } => {
-                metadata.partition_specs.insert(*spec.spec_id(), spec);
+                let spec_id = *spec.spec_id();
+                metadata.partition_specs.insert(spec_id, spec);
+                added_spec_id = Some(spec_id);
             }
             TableUpdate::SetDefaultSpec { spec_id } => {
-                metadata.default_spec_id = spec_id;
+                if spec_id == -1 {
+                    if let Some(added_spec_id) = added_spec_id {
+                        metadata.default_spec_id = added_spec_id;
+                    } else {
+                        return Err(Error::InvalidFormat(
+                            "Cannot set default spec to -1 without adding a spec first".to_string(),
+                        ));
+                    }
+                } else {
+                    metadata.default_spec_id = spec_id;
+                }
             }
             TableUpdate::AddSortOrder { sort_order } => {
-                metadata.sort_orders.insert(sort_order.order_id, sort_order);
+                let sort_order_id = sort_order.order_id;
+                metadata.sort_orders.insert(sort_order_id, sort_order);
+                added_sort_order_id = Some(sort_order_id);
             }
             TableUpdate::SetDefaultSortOrder { sort_order_id } => {
-                metadata.default_sort_order_id = sort_order_id;
+                if sort_order_id == -1 {
+                    if let Some(added_sort_order_id) = added_sort_order_id {
+                        metadata.default_sort_order_id = added_sort_order_id;
+                    } else {
+                        return Err(Error::InvalidFormat(
+                            "Cannot set default sort order to -1 without adding a sort order first"
+                                .to_string(),
+                        ));
+                    }
+                } else {
+                    metadata.default_sort_order_id = sort_order_id;
+                }
             }
             TableUpdate::AddSnapshot { snapshot } => {
                 metadata.snapshot_log.push(SnapshotLog {
@@ -489,6 +530,12 @@ pub fn apply_table_updates(
             }
         };
     }
+
+    // Lastly make sure `last-updated-ms` field is up-to-date
+    metadata.last_updated_ms = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as i64;
     Ok(())
 }
 

iceberg-rust/tests/overwrite_test.rs

@@ -83,6 +83,8 @@ async fn test_table_transaction_overwrite() {
         .await
         .expect("Failed to create table");
 
+    let mut previous_last_updated_ms = table.metadata().last_updated_ms;
+
     // 4. Create initial Arrow RecordBatch and write to parquet
     let initial_batch = create_initial_record_batch();
     let initial_stream = stream::iter(vec![Ok(initial_batch.clone())]);
@@ -104,6 +106,8 @@ async fn test_table_transaction_overwrite() {
         !table.metadata().snapshots.is_empty(),
         "Table should have at least one snapshot after append"
     );
+    assert!(table.metadata().last_updated_ms > previous_last_updated_ms);
+    previous_last_updated_ms = table.metadata().last_updated_ms;
 
     // 6. Create overwrite RecordBatch with additional rows
     let overwrite_batch = create_overwrite_record_batch();
@@ -134,6 +138,7 @@ async fn test_table_transaction_overwrite() {
         final_snapshots.len() >= 2,
         "Table should have at least 2 snapshots after overwrite"
     );
+    assert!(table.metadata().last_updated_ms > previous_last_updated_ms);
 
     // Get the current snapshot (should be the overwrite snapshot)
     let current_snapshot = table