Add support for evolving a partition column (#1334)

## Which issue does this PR close?

To check what happens if a partition column is evolved.

## What changes are included in this PR?

<!--
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?

<!--
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)?
-->

---------

Co-authored-by: Kevin Liu <kevinjqliu@users.noreply.github.com>

Author: Fokko

Cargo.lock

@@ -403,12 +403,26 @@ impl Datum {
                 }
             }
             PrimitiveType::Int => PrimitiveLiteral::Int(i32::from_le_bytes(bytes.try_into()?)),
-            PrimitiveType::Long => PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?)),
+            PrimitiveType::Long => {
+                if bytes.len() == 4 {
+                    // In the case of an evolved field
+                    PrimitiveLiteral::Long(i32::from_le_bytes(bytes.try_into()?) as i64)
+                } else {
+                    PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?))
+                }
+            }
             PrimitiveType::Float => {
                 PrimitiveLiteral::Float(OrderedFloat(f32::from_le_bytes(bytes.try_into()?)))
             }
             PrimitiveType::Double => {
-                PrimitiveLiteral::Double(OrderedFloat(f64::from_le_bytes(bytes.try_into()?)))
+                if bytes.len() == 4 {
+                    // In the case of an evolved field
+                    PrimitiveLiteral::Double(OrderedFloat(
+                        f32::from_le_bytes(bytes.try_into()?) as f64
+                    ))
+                } else {
+                    PrimitiveLiteral::Double(OrderedFloat(f64::from_le_bytes(bytes.try_into()?)))
+                }
             }
             PrimitiveType::Date => PrimitiveLiteral::Int(i32::from_le_bytes(bytes.try_into()?)),
             PrimitiveType::Time => PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?)),
@@ -3172,6 +3186,13 @@ mod tests {
         check_avro_bytes_serde(bytes, Datum::long(32), &PrimitiveType::Long);
     }
 
+    #[test]
+    fn avro_bytes_long_from_int() {
+        let bytes = vec![32u8, 0u8, 0u8, 0u8];
+
+        check_avro_bytes_serde(bytes, Datum::long(32), &PrimitiveType::Long);
+    }
+
     #[test]
     fn avro_bytes_float() {
         let bytes = vec![0u8, 0u8, 128u8, 63u8];
@@ -3186,6 +3207,13 @@ mod tests {
         check_avro_bytes_serde(bytes, Datum::double(1.0), &PrimitiveType::Double);
     }
 
+    #[test]
+    fn avro_bytes_double_from_float() {
+        let bytes = vec![0u8, 0u8, 128u8, 63u8];
+
+        check_avro_bytes_serde(bytes, Datum::double(1.0), &PrimitiveType::Double);
+    }
+
     #[test]
     fn avro_bytes_string() {
         let bytes = vec![105u8, 99u8, 101u8, 98u8, 101u8, 114u8, 103u8];

crates/iceberg/src/spec/values.rs

@@ -37,3 +37,4 @@ iceberg_test_utils = { path = "../test_utils", features = ["tests"] }
 parquet = { workspace = true }
 tokio = { workspace = true }
 uuid = { workspace = true }
+ordered-float = "2.10.1"

crates/integration_tests/Cargo.toml

@@ -120,6 +120,14 @@
 spark.sql("ALTER TABLE rest.default.test_promote_column ALTER COLUMN foo TYPE bigint")
 spark.sql("INSERT INTO rest.default.test_promote_column VALUES (25)")
 
+#  Create a table, and do some evolution on a partition column
+spark.sql("CREATE OR REPLACE TABLE rest.default.test_promote_partition_column (foo int, bar float, baz decimal(4, 2)) USING iceberg PARTITIONED BY (foo)")
+spark.sql("INSERT INTO rest.default.test_promote_partition_column VALUES (19, 19.25, 19.25)")
+spark.sql("ALTER TABLE rest.default.test_promote_partition_column ALTER COLUMN foo TYPE bigint")
+spark.sql("ALTER TABLE rest.default.test_promote_partition_column ALTER COLUMN bar TYPE double")
+spark.sql("ALTER TABLE rest.default.test_promote_partition_column ALTER COLUMN baz TYPE decimal(6, 2)")
+spark.sql("INSERT INTO rest.default.test_promote_partition_column VALUES (25, 22.25, 22.25)")
+
 #  Create a table with various types
 spark.sql("""
 CREATE OR REPLACE TABLE rest.default.types_test USING ICEBERG AS 

crates/integration_tests/testdata/spark/provision.py

@@ -17,12 +17,13 @@
 
 //! Integration tests for rest catalog.
 
-use arrow_array::{Int64Array, StringArray};
+use arrow_array::{Decimal128Array, Float64Array, Int64Array, StringArray};
 use futures::TryStreamExt;
 use iceberg::expr::Reference;
 use iceberg::spec::Datum;
 use iceberg::{Catalog, TableIdent};
 use iceberg_catalog_rest::RestCatalog;
+use ordered_float::OrderedFloat;
 
 use crate::get_shared_containers;
 
@@ -98,4 +99,80 @@ async fn test_evolved_schema() {
     actual.sort();
 
     assert_eq!(actual, vec![19, 25]);
+
+    // Evolve a partitioned column
+    let table = rest_catalog
+        .load_table(&TableIdent::from_strs(["default", "test_promote_partition_column"]).unwrap())
+        .await
+        .unwrap();
+    let scan = table.scan().build();
+    let batch_stream = scan.unwrap().to_arrow().await.unwrap();
+
+    let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
+    let mut actual_foo = vec![
+        batches[0]
+            .column_by_name("foo")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Int64Array>()
+            .unwrap()
+            .value(0),
+        batches[1]
+            .column_by_name("foo")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Int64Array>()
+            .unwrap()
+            .value(0),
+    ];
+
+    actual_foo.sort();
+
+    assert_eq!(actual_foo, vec![19, 25]);
+
+    let mut actual_bar = vec![
+        OrderedFloat(
+            batches[0]
+                .column_by_name("bar")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<Float64Array>()
+                .unwrap()
+                .value(0),
+        ),
+        OrderedFloat(
+            batches[1]
+                .column_by_name("bar")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<Float64Array>()
+                .unwrap()
+                .value(0),
+        ),
+    ];
+
+    actual_bar.sort();
+
+    assert_eq!(actual_bar, vec![19.25, 22.25]);
+
+    let mut actual_baz = vec![
+        batches[0]
+            .column_by_name("baz")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Decimal128Array>()
+            .unwrap()
+            .value(0),
+        batches[1]
+            .column_by_name("baz")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Decimal128Array>()
+            .unwrap()
+            .value(0),
+    ];
+
+    actual_baz.sort();
+
+    assert_eq!(actual_baz, vec![1925, 2225]);
 }