fix: table UUID partitions not working

crates/iceberg/src/spec/values/serde.rs

@@ -23,6 +23,7 @@ pub(crate) mod _serde {
     use serde::{Deserialize, Serialize};
     use serde_bytes::ByteBuf;
     use serde_derive::{Deserialize as DeserializeDerive, Serialize as SerializeDerive};
+    use uuid::Uuid;
 
     use crate::spec::values::{Literal, Map, PrimitiveLiteral, Struct};
     use crate::spec::{MAP_KEY_FIELD_NAME, MAP_VALUE_FIELD_NAME, PrimitiveType, Type};
@@ -238,7 +239,11 @@ pub(crate) mod _serde {
                     PrimitiveLiteral::Double(v) => RawLiteralEnum::Double(v.0),
                     PrimitiveLiteral::String(v) => RawLiteralEnum::String(v),
                     PrimitiveLiteral::UInt128(v) => {
-                        RawLiteralEnum::Bytes(ByteBuf::from(v.to_be_bytes()))
+                        if *ty == Type::Primitive(PrimitiveType::Uuid) {
+                            RawLiteralEnum::String(Uuid::from_u128(v).hyphenated().to_string())
+                        } else {
+                            RawLiteralEnum::Bytes(ByteBuf::from(v.to_be_bytes()))
+                        }
                     }
                     PrimitiveLiteral::Binary(v) => RawLiteralEnum::Bytes(ByteBuf::from(v)),
                     PrimitiveLiteral::Int128(v) => {
@@ -444,6 +449,12 @@ pub(crate) mod _serde {
                 },
                 RawLiteralEnum::String(v) => match ty {
                     Type::Primitive(PrimitiveType::String) => Ok(Some(Literal::string(v))),
+                    Type::Primitive(PrimitiveType::Uuid) => {
+                        let uuid = Uuid::parse_str(&v).map_err(|_| {
+                            invalid_err_with_reason("string", "UUID must be a valid UUID string")
+                        })?;
+                        Ok(Some(Literal::uuid(uuid)))
+                    }
                     _ => Err(invalid_err("string")),
                 },
                 RawLiteralEnum::Bytes(v) => match ty {
@@ -467,7 +478,7 @@ pub(crate) mod _serde {
                             let bytes: [u8; 16] = v.as_slice().try_into().map_err(|_| {
                                 invalid_err_with_reason("bytes", "UUID must be exactly 16 bytes")
                             })?;
-                            Ok(Some(Literal::uuid(uuid::Uuid::from_bytes(bytes))))
+                            Ok(Some(Literal::uuid(Uuid::from_bytes(bytes))))
                         } else {
                             Err(invalid_err_with_reason(
                                 "bytes",
@@ -601,7 +612,7 @@ pub(crate) mod _serde {
                                 ));
                             }
                         }
-                        Ok(Some(Literal::uuid(uuid::Uuid::from_bytes(bytes))))
+                        Ok(Some(Literal::uuid(Uuid::from_bytes(bytes))))
                     }
                     Type::Primitive(PrimitiveType::Decimal {
                         precision: _,

crates/iceberg/src/spec/values/tests.rs

@@ -240,7 +240,7 @@ fn json_timestamptz_ns_rejects_non_utc_offset() {
     // Per the spec, timestamptz_ns single-value serialization must use offset "+00:00"; Java's
     // SingleValueParser enforces the same (DateTimeUtil.isUTCTimestamptz). A non-UTC offset is not a
     // valid encoding and must be rejected, not silently re-based to UTC.
-    let record = serde_json::Value::String("2017-11-16T22:31:08.123456789+05:00".to_string());
+    let record = JsonValue::String("2017-11-16T22:31:08.123456789+05:00".to_string());
     let result = Literal::try_from_json(record, &Primitive(PrimitiveType::TimestamptzNs));
     assert!(
         result.is_err(),
@@ -252,7 +252,7 @@ fn json_timestamptz_ns_rejects_non_utc_offset() {
 fn json_timestamptz_rejects_non_utc_offset() {
     // Micros-precision counterpart, mirroring Java's TestSingleValueParser.testInvalidTimestamptz:
     // the offset must be "+00:00", so a non-UTC offset is rejected.
-    let record = serde_json::Value::String("2017-11-16T22:31:08.123456+05:00".to_string());
+    let record = JsonValue::String("2017-11-16T22:31:08.123456+05:00".to_string());
     let result = Literal::try_from_json(record, &Primitive(PrimitiveType::Timestamptz));
     assert!(
         result.is_err(),
@@ -529,20 +529,38 @@ fn check_raw_literal_bytes_serde_via_avro(
     expected_literal: Literal,
     expected_type: &Type,
 ) {
-    use apache_avro::types::Value;
-
     // Create an Avro bytes value and deserialize it through the RawLiteral path
     let avro_value = Value::Bytes(input_bytes);
-    let raw_literal: RawLiteral = apache_avro::from_value(&avro_value).unwrap();
-    let result = raw_literal.try_into(expected_type).unwrap();
-    assert_eq!(result, Some(expected_literal));
+    check_raw_literal_serde_via_avro(avro_value, expected_literal, expected_type);
 }
 
 fn check_raw_literal_bytes_error_via_avro(input_bytes: Vec<u8>, expected_type: &Type) {
-    use apache_avro::types::Value;
-
     let avro_value = Value::Bytes(input_bytes);
-    let raw_literal: RawLiteral = apache_avro::from_value(&avro_value).unwrap();
+    check_raw_literal_error_via_avro(avro_value, expected_type);
+}
+
+fn check_raw_literal_string_serde_via_avro(
+    input: &str,
+    expected_literal: Literal,
+    expected_type: &Type,
+) {
+    let avro_value = Value::String(input.to_string());
+    check_raw_literal_serde_via_avro(avro_value, expected_literal, expected_type);
+}
+
+fn check_raw_literal_string_error_via_avro(input: &str, expected_type: &Type) {
+    let avro_value = Value::String(input.to_string());
+    check_raw_literal_error_via_avro(avro_value, expected_type);
+}
+
+fn check_raw_literal_serde_via_avro(input: Value, expected_literal: Literal, expected_type: &Type) {
+    let raw_literal: RawLiteral = apache_avro::from_value(&input).unwrap();
+    let result = raw_literal.try_into(expected_type).unwrap();
+    assert_eq!(result, Some(expected_literal));
+}
+
+fn check_raw_literal_error_via_avro(input: Value, expected_type: &Type) {
+    let raw_literal: RawLiteral = apache_avro::from_value(&input).unwrap();
     let result = raw_literal.try_into(expected_type);
     assert!(result.is_err(), "Expected error but got: {result:?}");
 }
@@ -616,6 +634,21 @@ fn test_raw_literal_bytes_uuid_wrong_length() {
     check_raw_literal_bytes_error_via_avro(bytes, &Primitive(PrimitiveType::Uuid));
 }
 
+#[test]
+fn test_raw_literal_string_uuid_valid() {
+    let s = "f79c3e09-677c-4bbd-a479-3f349cb785e7";
+    check_raw_literal_string_serde_via_avro(
+        s,
+        Literal::uuid(Uuid::parse_str(s).unwrap()),
+        &Primitive(PrimitiveType::Uuid),
+    );
+}
+
+#[test]
+fn test_raw_literal_string_uuid_invalid() {
+    check_raw_literal_string_error_via_avro("not-a-uuid", &Primitive(PrimitiveType::Uuid));
+}
+
 #[test]
 fn test_raw_literal_bytes_decimal_precision_4_scale_2() {
     // Precision 4 requires 2 bytes

crates/integration_tests/src/lib.rs

@@ -16,12 +16,14 @@
 // under the License.
 
 use std::collections::HashMap;
-use std::sync::OnceLock;
+use std::sync::{Arc, OnceLock};
 
+use iceberg::CatalogBuilder;
 use iceberg::io::{
     S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_PATH_STYLE_ACCESS, S3_REGION, S3_SECRET_ACCESS_KEY,
 };
-use iceberg_catalog_rest::REST_CATALOG_PROP_URI;
+use iceberg_catalog_rest::{REST_CATALOG_PROP_URI, RestCatalog, RestCatalogBuilder};
+use iceberg_storage_opendal::OpenDalStorageFactory;
 use iceberg_test_utils::{get_minio_endpoint, get_rest_catalog_endpoint, set_up};
 
 /// Global test fixture that uses environment-based configuration.
@@ -52,6 +54,16 @@ impl GlobalTestFixture {
 
         GlobalTestFixture { catalog_config }
     }
+
+    pub async fn rest_catalog(&self) -> RestCatalog {
+        RestCatalogBuilder::default()
+            .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 {
+                customized_credential_load: None,
+            }))
+            .load("rest", self.catalog_config.clone())
+            .await
+            .unwrap()
+    }
 }
 
 /// Returns a reference to the global test fixture.

crates/integrations/datafusion/tests/integration_datafusion_test.rs

@@ -29,7 +29,8 @@ use expect_test::expect;
 use iceberg::io::LocalFsStorageFactory;
 use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
 use iceberg::spec::{
-    NestedField, PrimitiveType, Schema, StructType, Transform, Type, UnboundPartitionSpec,
+    NestedField, PrimitiveType, Schema, StructType, Transform, Type, UnboundPartitionField,
+    UnboundPartitionSpec,
 };
 use iceberg::test_utils::check_record_batches;
 use iceberg::{
@@ -601,8 +602,8 @@ async fn test_insert_into_nested() -> Result<()> {
     // Insert data with nested structs
     let insert_sql = r#"
     INSERT INTO catalog.test_insert_nested.nested_table
-    SELECT 
-        1 as id, 
+    SELECT
+        1 as id,
         'Alice' as name,
         named_struct(
             'address', named_struct(
@@ -616,8 +617,8 @@ async fn test_insert_into_nested() -> Result<()> {
             )
         ) as profile
     UNION ALL
-    SELECT 
-        2 as id, 
+    SELECT
+        2 as id,
         'Bob' as name,
         named_struct(
             'address', named_struct(
@@ -739,15 +740,15 @@ async fn test_insert_into_nested() -> Result<()> {
     let df = ctx
         .sql(
             r#"
-            SELECT 
-                id, 
+            SELECT
+                id,
                 name,
                 profile.address.street,
                 profile.address.city,
                 profile.address.zip,
                 profile.contact.email,
                 profile.contact.phone
-            FROM catalog.test_insert_nested.nested_table 
+            FROM catalog.test_insert_nested.nested_table
             ORDER BY id
         "#,
         )
@@ -811,7 +812,7 @@ async fn test_insert_into_nested() -> Result<()> {
 }
 
 #[tokio::test]
-async fn test_insert_into_partitioned() -> Result<()> {
+async fn test_insert_into_partitioned_by_string() -> Result<()> {
     let iceberg_catalog = get_iceberg_catalog().await;
     let namespace = NamespaceIdent::new("test_partitioned_write".to_string());
     set_test_namespace(&iceberg_catalog, &namespace).await?;
@@ -853,8 +854,8 @@ async fn test_insert_into_partitioned() -> Result<()> {
     let df = ctx
         .sql(
             r#"
-            INSERT INTO catalog.test_partitioned_write.partitioned_table 
-            VALUES 
+            INSERT INTO catalog.test_partitioned_write.partitioned_table
+            VALUES
                 (1, 'electronics', 'laptop'),
                 (2, 'electronics', 'phone'),
                 (3, 'books', 'novel'),
@@ -946,3 +947,98 @@ async fn test_insert_into_partitioned() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_insert_into_partitioned_by_uuid() -> Result<()> {
+    let iceberg_catalog = get_iceberg_catalog().await;
+    let namespace = NamespaceIdent::new("test_insert_uuid".to_string());
+    set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+    let schema = Schema::builder()
+        .with_schema_id(0)
+        .with_fields(vec![
+            NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+            NestedField::required(2, "uuid", Type::Primitive(PrimitiveType::Uuid)).into(),
+        ])
+        .build()?;
+
+    let creation = TableCreation::builder()
+        .location(temp_path())
+        .name("partitioned_table".to_string())
+        .properties(HashMap::new())
+        .schema(schema)
+        .partition_spec(
+            UnboundPartitionSpec::builder()
+                .with_spec_id(0)
+                .add_partition_fields([UnboundPartitionField::builder()
+                    .source_id(2)
+                    .field_id(2)
+                    .name("uuid".to_string())
+                    .transform(Transform::Identity)
+                    .build()])
+                .unwrap()
+                .build(),
+        )
+        .build();
+    iceberg_catalog.create_table(&namespace, creation).await?;
+
+    let client = Arc::new(iceberg_catalog);
+    let catalog = Arc::new(IcebergCatalogProvider::try_new(client.clone()).await?);
+    let ctx = SessionContext::new();
+    ctx.register_catalog("catalog", catalog);
+
+    ctx.sql(
+        "INSERT INTO catalog.test_insert_uuid.partitioned_table
+         VALUES (1, X'aaaaaaaabbbbccccddddeeeeeeeeeeee')",
+    )
+    .await
+    .unwrap()
+    .collect()
+    .await
+    .unwrap();
+
+    // Read the UUID back out.
+    let batches = ctx
+        .sql("SELECT * FROM catalog.test_insert_uuid.partitioned_table")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    check_record_batches(
+        batches,
+        expect![[r#"
+            Field { "id": Int32, metadata: {"PARQUET:field_id": "1"} },
+            Field { "uuid": FixedSizeBinary(16), metadata: {"PARQUET:field_id": "2"} }"#]],
+        expect![[r#"
+            id: PrimitiveArray<Int32>
+            [
+              1,
+            ],
+            uuid: FixedSizeBinaryArray<16>
+            [
+              [170, 170, 170, 170, 187, 187, 204, 204, 221, 221, 238, 238, 238, 238, 238, 238],
+            ]"#]],
+        &[],
+        Some("id"),
+    );
+
+    // Verify that data files exist under correct UUID paths
+    let table_ident = TableIdent::new(namespace.clone(), "partitioned_table".to_string());
+    let table = client.load_table(&table_ident).await?;
+    let table_location = table.metadata().location();
+    let file_io = table.file_io();
+
+    // List files under each expected partition path
+    let uuid_partition_path =
+        format!("{table_location}/data/uuid=aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee");
+
+    // Verify partition directories exist and contain data files
+    assert!(
+        file_io.exists(&uuid_partition_path).await?,
+        "Expected partition directory: {uuid_partition_path}"
+    );
+
+    Ok(())
+}