diff --git a/crates/iceberg/src/spec/values/serde.rs b/crates/iceberg/src/spec/values/serde.rs index 053acca8b0..a9fc381874 100644 --- a/crates/iceberg/src/spec/values/serde.rs +++ b/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: _, diff --git a/crates/iceberg/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index 8bf3311004..57909269f2 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/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, 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 diff --git a/crates/integration_tests/src/lib.rs b/crates/integration_tests/src/lib.rs index feafa3ae9f..e78b8544c5 100644 --- a/crates/integration_tests/src/lib.rs +++ b/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. diff --git a/crates/integrations/datafusion/tests/integration_datafusion_test.rs b/crates/integrations/datafusion/tests/integration_datafusion_test.rs index 9aa6a1a501..dd17afd721 100644 --- a/crates/integrations/datafusion/tests/integration_datafusion_test.rs +++ b/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 + [ + 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(()) +}