From ea998f952fcd83aa980803916052442d646de704 Mon Sep 17 00:00:00 2001 From: "josephlenton@gmail.com" Date: Mon, 27 Jul 2026 16:36:29 +0100 Subject: [PATCH 1/4] chore: add a test that recreates the uuid partition error --- .../tests/write_to_iceberg.rs | 247 ++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 crates/integration_tests/tests/write_to_iceberg.rs diff --git a/crates/integration_tests/tests/write_to_iceberg.rs b/crates/integration_tests/tests/write_to_iceberg.rs new file mode 100644 index 0000000000..22434dda4e --- /dev/null +++ b/crates/integration_tests/tests/write_to_iceberg.rs @@ -0,0 +1,247 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Integration tests for rest catalog. + +mod common; + +use std::sync::Arc; + +use arrow_array::{ + ArrayRef, BooleanArray, FixedSizeBinaryArray, Int32Array, RecordBatch, StringArray, +}; +use common::{random_ns, test_schema}; +use futures::TryStreamExt; +use iceberg::spec::{ + Literal, NestedField, PartitionKey, PrimitiveType, Schema, Struct, Transform, Type, + UnboundPartitionField, UnboundPartitionSpec, +}; +use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; +use iceberg::writer::file_writer::ParquetWriterBuilder; +use iceberg::writer::file_writer::location_generator::{ + DefaultFileNameGenerator, DefaultLocationGenerator, +}; +use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; +use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; +use iceberg::{Catalog, CatalogBuilder, TableCreation}; +use iceberg_catalog_rest::RestCatalogBuilder; +use iceberg_integration_tests::get_test_fixture; +use iceberg_storage_opendal::OpenDalStorageFactory; +use parquet::file::properties::WriterProperties; +use uuid::Uuid; + +#[tokio::test] +async fn it_should_write_a_table_with_uuid_fields() { + let fixture = get_test_fixture(); + let rest_catalog = RestCatalogBuilder::default() + .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: None, + })) + .load("rest", fixture.catalog_config.clone()) + .await + .unwrap(); + + let schema = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "uuid", Type::Primitive(PrimitiveType::Uuid)).into(), + ]) + .build() + .unwrap(); + + let table_creation = TableCreation::builder() + .name("t1".to_string()) + .schema(schema.clone()) + .build(); + + let ns = random_ns().await; + let table = rest_catalog + .create_table(ns.name(), table_creation) + .await + .unwrap(); + + // Create the writer and write the data + let schema: Arc = Arc::new( + table + .metadata() + .current_schema() + .as_ref() + .try_into() + .unwrap(), + ); + let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap(); + let file_name_generator = DefaultFileNameGenerator::new( + "test".to_string(), + None, + iceberg::spec::DataFileFormat::Parquet, + ); + let parquet_writer_builder = ParquetWriterBuilder::new( + WriterProperties::default(), + table.metadata().current_schema().clone(), + ); + let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + table.file_io().clone(), + location_generator.clone(), + file_name_generator.clone(), + ); + let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); + let mut data_file_writer = data_file_writer_builder.build(None).await.unwrap(); + let uuid_value = Uuid::from_u128(0x01234567_0000_0000_0000_000000000000); + let col = FixedSizeBinaryArray::try_from_iter(std::iter::once(uuid_value.as_bytes().to_vec())) + .unwrap(); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(col) as ArrayRef]).unwrap(); + data_file_writer.write(batch.clone()).await.unwrap(); + let data_file = data_file_writer.close().await.unwrap(); + + // start two transaction and commit one of them + let tx1 = Transaction::new(&table); + let append_action = tx1.fast_append().add_data_files(data_file.clone()); + let tx1 = append_action.apply(tx1).unwrap(); + + let table = tx1 + .commit(&rest_catalog) + .await + .expect("The first commit should not fail."); + + // check result + let batch_stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0], batch); +} + +#[tokio::test] +async fn it_should_write_a_table_with_uuid_fields_and_uuid_partition() { + let fixture = get_test_fixture(); + let rest_catalog = RestCatalogBuilder::default() + .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: None, + })) + .load("rest", fixture.catalog_config.clone()) + .await + .unwrap(); + + let schema = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "uuid", Type::Primitive(PrimitiveType::Uuid)).into(), + ]) + .build() + .unwrap(); + + let table_creation = TableCreation::builder() + .name("t1".to_string()) + .partition_spec( + UnboundPartitionSpec::builder() + .with_spec_id(0) + .add_partition_fields([UnboundPartitionField::builder() + .source_id(1) + .field_id(1) + .name("uuid".to_string()) + .transform(Transform::Identity) + .build()]) + .unwrap() + .build(), + ) + .schema(schema.clone()) + .build(); + + let ns = random_ns().await; + let table = rest_catalog + .create_table(ns.name(), table_creation) + .await + .unwrap(); + + // Create the writer and write the data + let arrow_schema: Arc = Arc::new( + table + .metadata() + .current_schema() + .as_ref() + .try_into() + .unwrap(), + ); + let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap(); + let file_name_generator = DefaultFileNameGenerator::new( + "test".to_string(), + None, + iceberg::spec::DataFileFormat::Parquet, + ); + let parquet_writer_builder = ParquetWriterBuilder::new( + WriterProperties::default(), + table.metadata().current_schema().clone(), + ); + let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + table.file_io().clone(), + location_generator.clone(), + file_name_generator.clone(), + ); + let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); + let uuid_value = Uuid::from_u128(0x01234567_0000_0000_0000_000000000000); + + let table_metadata = table.metadata(); + let spec = table_metadata.default_partition_spec().as_ref().clone(); + let table_schema = table_metadata.current_schema().clone(); + + let uuid_key = Some(Literal::uuid(uuid_value)); + let data = Struct::from_iter([uuid_key]); + let partition_key = PartitionKey::new(spec, table_schema.clone(), data); + + let mut data_file_writer = data_file_writer_builder + .build(Some(partition_key)) + .await + .unwrap(); + let col = FixedSizeBinaryArray::try_from_iter(std::iter::once(uuid_value.as_bytes().to_vec())) + .unwrap(); + let batch = + RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(col) as ArrayRef]).unwrap(); + data_file_writer.write(batch.clone()).await.unwrap(); + let data_file = data_file_writer.close().await.unwrap(); + + // start two transaction and commit one of them + let tx1 = Transaction::new(&table); + let append_action = tx1.fast_append().add_data_files(data_file.clone()); + let tx1 = append_action.apply(tx1).unwrap(); + + let table = tx1 + .commit(&rest_catalog) + .await + .expect("The first commit should not fail."); + + // check result + let batch_stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0], batch); +} From a7d205cd1b6623d90d0bf5b8fd3b9028dd08a2c1 Mon Sep 17 00:00:00 2001 From: "josephlenton@gmail.com" Date: Mon, 27 Jul 2026 20:17:32 +0100 Subject: [PATCH 2/4] fix: update to turn UUIDs into Strings for Partitions --- crates/iceberg/src/spec/values/serde.rs | 17 +- crates/integration_tests/src/lib.rs | 16 +- .../integration_tests/tests/write_to_table.rs | 206 ++++++++++++++++++ ...g.rs => write_to_table_with_partitions.rs} | 121 +--------- 4 files changed, 245 insertions(+), 115 deletions(-) create mode 100644 crates/integration_tests/tests/write_to_table.rs rename crates/integration_tests/tests/{write_to_iceberg.rs => write_to_table_with_partitions.rs} (54%) 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/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/integration_tests/tests/write_to_table.rs b/crates/integration_tests/tests/write_to_table.rs new file mode 100644 index 0000000000..f78724bd9e --- /dev/null +++ b/crates/integration_tests/tests/write_to_table.rs @@ -0,0 +1,206 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Integration tests for rest catalog. + +mod common; + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow_array::{ + ArrayRef, BooleanArray, Date32Array, Decimal128Array, FixedSizeBinaryArray, Float32Array, + Float64Array, Int32Array, Int64Array, LargeBinaryArray, RecordBatch, StringArray, + Time64MicrosecondArray, TimestampMicrosecondArray, TimestampNanosecondArray, +}; +use common::random_ns; +use futures::TryStreamExt; +use iceberg::spec::{NestedField, PrimitiveType, Schema, TableProperties, Type}; +use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; +use iceberg::writer::file_writer::ParquetWriterBuilder; +use iceberg::writer::file_writer::location_generator::{ + DefaultFileNameGenerator, DefaultLocationGenerator, +}; +use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; +use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; +use iceberg::{Catalog, TableCreation}; +use iceberg_integration_tests::get_test_fixture; +use parquet::file::properties::WriterProperties; +use uuid::Uuid; + +#[tokio::test] +async fn test_writing_to_a_table_with_all_primitive_types() { + let fixture = get_test_fixture(); + let rest_catalog = fixture.rest_catalog().await; + + let schema = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "boolean", Type::Primitive(PrimitiveType::Boolean)).into(), + NestedField::required(2, "int", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(3, "long", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(4, "float", Type::Primitive(PrimitiveType::Float)).into(), + NestedField::required(5, "double", Type::Primitive(PrimitiveType::Double)).into(), + NestedField::required( + 6, + "decimal", + Type::Primitive(PrimitiveType::Decimal { + precision: 38, + scale: 10, + }), + ) + .into(), + NestedField::required(7, "date", Type::Primitive(PrimitiveType::Date)).into(), + NestedField::required(8, "time", Type::Primitive(PrimitiveType::Time)).into(), + NestedField::required(9, "timestamp", Type::Primitive(PrimitiveType::Timestamp)).into(), + NestedField::required( + 10, + "timestamptz", + Type::Primitive(PrimitiveType::Timestamptz), + ) + .into(), + NestedField::required( + 11, + "timestamp_ns", + Type::Primitive(PrimitiveType::TimestampNs), + ) + .into(), + NestedField::required( + 12, + "timestamptz_ns", + Type::Primitive(PrimitiveType::TimestamptzNs), + ) + .into(), + NestedField::required(13, "string", Type::Primitive(PrimitiveType::String)).into(), + NestedField::required(14, "uuid", Type::Primitive(PrimitiveType::Uuid)).into(), + NestedField::required(15, "fixed", Type::Primitive(PrimitiveType::Fixed(16))).into(), + NestedField::required(16, "binary", Type::Primitive(PrimitiveType::Binary)).into(), + ]) + .build() + .unwrap(); + + let table_creation = TableCreation::builder() + .name("t1".to_string()) + .schema(schema.clone()) + // for timestamptz_ns support + .properties(HashMap::::from_iter([( + TableProperties::PROPERTY_FORMAT_VERSION.to_string(), + "3".to_string(), + )])) + .build(); + + let ns = random_ns().await; + let table = rest_catalog + .create_table(ns.name(), table_creation) + .await + .unwrap(); + + // Create the writer and write the data + let schema = Arc::::new( + table + .metadata() + .current_schema() + .as_ref() + .try_into() + .unwrap(), + ); + let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap(); + let file_name_generator = DefaultFileNameGenerator::new( + "test".to_string(), + None, + iceberg::spec::DataFileFormat::Parquet, + ); + let parquet_writer_builder = ParquetWriterBuilder::new( + WriterProperties::default(), + table.metadata().current_schema().clone(), + ); + let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + table.file_io().clone(), + location_generator.clone(), + file_name_generator.clone(), + ); + let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); + let mut data_file_writer = data_file_writer_builder.build(None).await.unwrap(); + let batch = RecordBatch::try_new(schema.clone(), vec![ + Arc::new(BooleanArray::from(vec![true])) as ArrayRef, + Arc::new(Int32Array::from(vec![42])) as ArrayRef, + Arc::new(Int64Array::from(vec![42i64])) as ArrayRef, + Arc::new(Float32Array::from(vec![1.5f32])) as ArrayRef, + Arc::new(Float64Array::from(vec![2.5f64])) as ArrayRef, + Arc::new( + Decimal128Array::from(vec![12345678901234567890i128]) + .with_precision_and_scale(38, 10) + .unwrap(), + ) as ArrayRef, + Arc::new(Date32Array::from(vec![19000])) as ArrayRef, + Arc::new(Time64MicrosecondArray::from(vec![123456789i64])) as ArrayRef, + Arc::new(TimestampMicrosecondArray::from(vec![ + 1_600_000_000_000_000i64, + ])) as ArrayRef, + Arc::new( + TimestampMicrosecondArray::from(vec![1_600_000_000_000_000i64]).with_timezone("+00:00"), + ) as ArrayRef, + Arc::new(TimestampNanosecondArray::from(vec![ + 1_600_000_000_000_000_000i64, + ])) as ArrayRef, + Arc::new( + TimestampNanosecondArray::from(vec![1_600_000_000_000_000_000i64]) + .with_timezone("+00:00"), + ) as ArrayRef, + Arc::new(StringArray::from(vec!["🦊"])) as ArrayRef, + Arc::new( + FixedSizeBinaryArray::try_from_iter(std::iter::once( + Uuid::from_u128(0xa1a2a3a4b1b2c1c2d1d2d3d4d5d6d7d8u128) + .as_bytes() + .to_vec(), + )) + .unwrap(), + ) as ArrayRef, + Arc::new(FixedSizeBinaryArray::try_from_iter(std::iter::once(vec![0u8; 16])).unwrap()) + as ArrayRef, + Arc::new(LargeBinaryArray::from_iter_values(std::iter::once( + b"binary".as_slice(), + ))) as ArrayRef, + ]) + .unwrap(); + data_file_writer.write(batch.clone()).await.unwrap(); + let data_file = data_file_writer.close().await.unwrap(); + + let tx = Transaction::new(&table); + let append_action = tx.fast_append().add_data_files(data_file.clone()); + let tx = append_action.apply(tx).unwrap(); + + let table = tx + .commit(&rest_catalog) + .await + .expect("The first commit should not fail."); + + // check results + let batch_stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0], batch); +} diff --git a/crates/integration_tests/tests/write_to_iceberg.rs b/crates/integration_tests/tests/write_to_table_with_partitions.rs similarity index 54% rename from crates/integration_tests/tests/write_to_iceberg.rs rename to crates/integration_tests/tests/write_to_table_with_partitions.rs index 22434dda4e..f8034949c9 100644 --- a/crates/integration_tests/tests/write_to_iceberg.rs +++ b/crates/integration_tests/tests/write_to_table_with_partitions.rs @@ -21,10 +21,8 @@ mod common; use std::sync::Arc; -use arrow_array::{ - ArrayRef, BooleanArray, FixedSizeBinaryArray, Int32Array, RecordBatch, StringArray, -}; -use common::{random_ns, test_schema}; +use arrow_array::{ArrayRef, FixedSizeBinaryArray, RecordBatch}; +use common::random_ns; use futures::TryStreamExt; use iceberg::spec::{ Literal, NestedField, PartitionKey, PrimitiveType, Schema, Struct, Transform, Type, @@ -38,111 +36,15 @@ use iceberg::writer::file_writer::location_generator::{ }; use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; -use iceberg::{Catalog, CatalogBuilder, TableCreation}; -use iceberg_catalog_rest::RestCatalogBuilder; +use iceberg::{Catalog, TableCreation}; use iceberg_integration_tests::get_test_fixture; -use iceberg_storage_opendal::OpenDalStorageFactory; use parquet::file::properties::WriterProperties; use uuid::Uuid; #[tokio::test] -async fn it_should_write_a_table_with_uuid_fields() { +async fn test_writing_to_a_table_with_uuid_partition() { let fixture = get_test_fixture(); - let rest_catalog = RestCatalogBuilder::default() - .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { - customized_credential_load: None, - })) - .load("rest", fixture.catalog_config.clone()) - .await - .unwrap(); - - let schema = Schema::builder() - .with_schema_id(1) - .with_fields(vec![ - NestedField::required(1, "uuid", Type::Primitive(PrimitiveType::Uuid)).into(), - ]) - .build() - .unwrap(); - - let table_creation = TableCreation::builder() - .name("t1".to_string()) - .schema(schema.clone()) - .build(); - - let ns = random_ns().await; - let table = rest_catalog - .create_table(ns.name(), table_creation) - .await - .unwrap(); - - // Create the writer and write the data - let schema: Arc = Arc::new( - table - .metadata() - .current_schema() - .as_ref() - .try_into() - .unwrap(), - ); - let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap(); - let file_name_generator = DefaultFileNameGenerator::new( - "test".to_string(), - None, - iceberg::spec::DataFileFormat::Parquet, - ); - let parquet_writer_builder = ParquetWriterBuilder::new( - WriterProperties::default(), - table.metadata().current_schema().clone(), - ); - let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( - parquet_writer_builder, - table.file_io().clone(), - location_generator.clone(), - file_name_generator.clone(), - ); - let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); - let mut data_file_writer = data_file_writer_builder.build(None).await.unwrap(); - let uuid_value = Uuid::from_u128(0x01234567_0000_0000_0000_000000000000); - let col = FixedSizeBinaryArray::try_from_iter(std::iter::once(uuid_value.as_bytes().to_vec())) - .unwrap(); - let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(col) as ArrayRef]).unwrap(); - data_file_writer.write(batch.clone()).await.unwrap(); - let data_file = data_file_writer.close().await.unwrap(); - - // start two transaction and commit one of them - let tx1 = Transaction::new(&table); - let append_action = tx1.fast_append().add_data_files(data_file.clone()); - let tx1 = append_action.apply(tx1).unwrap(); - - let table = tx1 - .commit(&rest_catalog) - .await - .expect("The first commit should not fail."); - - // check result - let batch_stream = table - .scan() - .select_all() - .build() - .unwrap() - .to_arrow() - .await - .unwrap(); - let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); - assert_eq!(batches.len(), 1); - assert_eq!(batches[0], batch); -} - -#[tokio::test] -async fn it_should_write_a_table_with_uuid_fields_and_uuid_partition() { - let fixture = get_test_fixture(); - let rest_catalog = RestCatalogBuilder::default() - .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { - customized_credential_load: None, - })) - .load("rest", fixture.catalog_config.clone()) - .await - .unwrap(); + let rest_catalog = fixture.rest_catalog().await; let schema = Schema::builder() .with_schema_id(1) @@ -201,12 +103,12 @@ async fn it_should_write_a_table_with_uuid_fields_and_uuid_partition() { file_name_generator.clone(), ); let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); - let uuid_value = Uuid::from_u128(0x01234567_0000_0000_0000_000000000000); let table_metadata = table.metadata(); let spec = table_metadata.default_partition_spec().as_ref().clone(); let table_schema = table_metadata.current_schema().clone(); + let uuid_value = Uuid::from_u128(0xa1a2a3a4b1b2c1c2d1d2d3d4d5d6d7d8u128); let uuid_key = Some(Literal::uuid(uuid_value)); let data = Struct::from_iter([uuid_key]); let partition_key = PartitionKey::new(spec, table_schema.clone(), data); @@ -222,17 +124,16 @@ async fn it_should_write_a_table_with_uuid_fields_and_uuid_partition() { data_file_writer.write(batch.clone()).await.unwrap(); let data_file = data_file_writer.close().await.unwrap(); - // start two transaction and commit one of them - let tx1 = Transaction::new(&table); - let append_action = tx1.fast_append().add_data_files(data_file.clone()); - let tx1 = append_action.apply(tx1).unwrap(); + let tx = Transaction::new(&table); + let append_action = tx.fast_append().add_data_files(data_file.clone()); + let tx = append_action.apply(tx).unwrap(); - let table = tx1 + let table = tx .commit(&rest_catalog) .await .expect("The first commit should not fail."); - // check result + // check results let batch_stream = table .scan() .select_all() From 490975252ac8f3cf5414781be9e14294fa8a7dac Mon Sep 17 00:00:00 2001 From: "josephlenton@gmail.com" Date: Tue, 28 Jul 2026 15:44:02 +0100 Subject: [PATCH 3/4] chore: add a unit test for RawLiteralEnum::try_from with uuid --- crates/iceberg/src/spec/values/tests.rs | 53 ++++++++++++++++++++----- 1 file changed, 43 insertions(+), 10 deletions(-) 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 From 6ca9fa26d1e462af7d3898d197eae4f7c9f2ae47 Mon Sep 17 00:00:00 2001 From: "josephlenton@gmail.com" Date: Wed, 29 Jul 2026 11:00:37 +0100 Subject: [PATCH 4/4] chore: move UUID partition test to use datafusion integration tests --- .../integration_tests/tests/write_to_table.rs | 206 ------------------ .../tests/write_to_table_with_partitions.rs | 148 ------------- .../tests/integration_datafusion_test.rs | 118 +++++++++- 3 files changed, 107 insertions(+), 365 deletions(-) delete mode 100644 crates/integration_tests/tests/write_to_table.rs delete mode 100644 crates/integration_tests/tests/write_to_table_with_partitions.rs diff --git a/crates/integration_tests/tests/write_to_table.rs b/crates/integration_tests/tests/write_to_table.rs deleted file mode 100644 index f78724bd9e..0000000000 --- a/crates/integration_tests/tests/write_to_table.rs +++ /dev/null @@ -1,206 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -//! Integration tests for rest catalog. - -mod common; - -use std::collections::HashMap; -use std::sync::Arc; - -use arrow_array::{ - ArrayRef, BooleanArray, Date32Array, Decimal128Array, FixedSizeBinaryArray, Float32Array, - Float64Array, Int32Array, Int64Array, LargeBinaryArray, RecordBatch, StringArray, - Time64MicrosecondArray, TimestampMicrosecondArray, TimestampNanosecondArray, -}; -use common::random_ns; -use futures::TryStreamExt; -use iceberg::spec::{NestedField, PrimitiveType, Schema, TableProperties, Type}; -use iceberg::transaction::{ApplyTransactionAction, Transaction}; -use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; -use iceberg::writer::file_writer::ParquetWriterBuilder; -use iceberg::writer::file_writer::location_generator::{ - DefaultFileNameGenerator, DefaultLocationGenerator, -}; -use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; -use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; -use iceberg::{Catalog, TableCreation}; -use iceberg_integration_tests::get_test_fixture; -use parquet::file::properties::WriterProperties; -use uuid::Uuid; - -#[tokio::test] -async fn test_writing_to_a_table_with_all_primitive_types() { - let fixture = get_test_fixture(); - let rest_catalog = fixture.rest_catalog().await; - - let schema = Schema::builder() - .with_schema_id(1) - .with_fields(vec![ - NestedField::required(1, "boolean", Type::Primitive(PrimitiveType::Boolean)).into(), - NestedField::required(2, "int", Type::Primitive(PrimitiveType::Int)).into(), - NestedField::required(3, "long", Type::Primitive(PrimitiveType::Long)).into(), - NestedField::required(4, "float", Type::Primitive(PrimitiveType::Float)).into(), - NestedField::required(5, "double", Type::Primitive(PrimitiveType::Double)).into(), - NestedField::required( - 6, - "decimal", - Type::Primitive(PrimitiveType::Decimal { - precision: 38, - scale: 10, - }), - ) - .into(), - NestedField::required(7, "date", Type::Primitive(PrimitiveType::Date)).into(), - NestedField::required(8, "time", Type::Primitive(PrimitiveType::Time)).into(), - NestedField::required(9, "timestamp", Type::Primitive(PrimitiveType::Timestamp)).into(), - NestedField::required( - 10, - "timestamptz", - Type::Primitive(PrimitiveType::Timestamptz), - ) - .into(), - NestedField::required( - 11, - "timestamp_ns", - Type::Primitive(PrimitiveType::TimestampNs), - ) - .into(), - NestedField::required( - 12, - "timestamptz_ns", - Type::Primitive(PrimitiveType::TimestamptzNs), - ) - .into(), - NestedField::required(13, "string", Type::Primitive(PrimitiveType::String)).into(), - NestedField::required(14, "uuid", Type::Primitive(PrimitiveType::Uuid)).into(), - NestedField::required(15, "fixed", Type::Primitive(PrimitiveType::Fixed(16))).into(), - NestedField::required(16, "binary", Type::Primitive(PrimitiveType::Binary)).into(), - ]) - .build() - .unwrap(); - - let table_creation = TableCreation::builder() - .name("t1".to_string()) - .schema(schema.clone()) - // for timestamptz_ns support - .properties(HashMap::::from_iter([( - TableProperties::PROPERTY_FORMAT_VERSION.to_string(), - "3".to_string(), - )])) - .build(); - - let ns = random_ns().await; - let table = rest_catalog - .create_table(ns.name(), table_creation) - .await - .unwrap(); - - // Create the writer and write the data - let schema = Arc::::new( - table - .metadata() - .current_schema() - .as_ref() - .try_into() - .unwrap(), - ); - let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap(); - let file_name_generator = DefaultFileNameGenerator::new( - "test".to_string(), - None, - iceberg::spec::DataFileFormat::Parquet, - ); - let parquet_writer_builder = ParquetWriterBuilder::new( - WriterProperties::default(), - table.metadata().current_schema().clone(), - ); - let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( - parquet_writer_builder, - table.file_io().clone(), - location_generator.clone(), - file_name_generator.clone(), - ); - let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); - let mut data_file_writer = data_file_writer_builder.build(None).await.unwrap(); - let batch = RecordBatch::try_new(schema.clone(), vec![ - Arc::new(BooleanArray::from(vec![true])) as ArrayRef, - Arc::new(Int32Array::from(vec![42])) as ArrayRef, - Arc::new(Int64Array::from(vec![42i64])) as ArrayRef, - Arc::new(Float32Array::from(vec![1.5f32])) as ArrayRef, - Arc::new(Float64Array::from(vec![2.5f64])) as ArrayRef, - Arc::new( - Decimal128Array::from(vec![12345678901234567890i128]) - .with_precision_and_scale(38, 10) - .unwrap(), - ) as ArrayRef, - Arc::new(Date32Array::from(vec![19000])) as ArrayRef, - Arc::new(Time64MicrosecondArray::from(vec![123456789i64])) as ArrayRef, - Arc::new(TimestampMicrosecondArray::from(vec![ - 1_600_000_000_000_000i64, - ])) as ArrayRef, - Arc::new( - TimestampMicrosecondArray::from(vec![1_600_000_000_000_000i64]).with_timezone("+00:00"), - ) as ArrayRef, - Arc::new(TimestampNanosecondArray::from(vec![ - 1_600_000_000_000_000_000i64, - ])) as ArrayRef, - Arc::new( - TimestampNanosecondArray::from(vec![1_600_000_000_000_000_000i64]) - .with_timezone("+00:00"), - ) as ArrayRef, - Arc::new(StringArray::from(vec!["🦊"])) as ArrayRef, - Arc::new( - FixedSizeBinaryArray::try_from_iter(std::iter::once( - Uuid::from_u128(0xa1a2a3a4b1b2c1c2d1d2d3d4d5d6d7d8u128) - .as_bytes() - .to_vec(), - )) - .unwrap(), - ) as ArrayRef, - Arc::new(FixedSizeBinaryArray::try_from_iter(std::iter::once(vec![0u8; 16])).unwrap()) - as ArrayRef, - Arc::new(LargeBinaryArray::from_iter_values(std::iter::once( - b"binary".as_slice(), - ))) as ArrayRef, - ]) - .unwrap(); - data_file_writer.write(batch.clone()).await.unwrap(); - let data_file = data_file_writer.close().await.unwrap(); - - let tx = Transaction::new(&table); - let append_action = tx.fast_append().add_data_files(data_file.clone()); - let tx = append_action.apply(tx).unwrap(); - - let table = tx - .commit(&rest_catalog) - .await - .expect("The first commit should not fail."); - - // check results - let batch_stream = table - .scan() - .select_all() - .build() - .unwrap() - .to_arrow() - .await - .unwrap(); - let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); - assert_eq!(batches.len(), 1); - assert_eq!(batches[0], batch); -} diff --git a/crates/integration_tests/tests/write_to_table_with_partitions.rs b/crates/integration_tests/tests/write_to_table_with_partitions.rs deleted file mode 100644 index f8034949c9..0000000000 --- a/crates/integration_tests/tests/write_to_table_with_partitions.rs +++ /dev/null @@ -1,148 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -//! Integration tests for rest catalog. - -mod common; - -use std::sync::Arc; - -use arrow_array::{ArrayRef, FixedSizeBinaryArray, RecordBatch}; -use common::random_ns; -use futures::TryStreamExt; -use iceberg::spec::{ - Literal, NestedField, PartitionKey, PrimitiveType, Schema, Struct, Transform, Type, - UnboundPartitionField, UnboundPartitionSpec, -}; -use iceberg::transaction::{ApplyTransactionAction, Transaction}; -use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; -use iceberg::writer::file_writer::ParquetWriterBuilder; -use iceberg::writer::file_writer::location_generator::{ - DefaultFileNameGenerator, DefaultLocationGenerator, -}; -use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; -use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; -use iceberg::{Catalog, TableCreation}; -use iceberg_integration_tests::get_test_fixture; -use parquet::file::properties::WriterProperties; -use uuid::Uuid; - -#[tokio::test] -async fn test_writing_to_a_table_with_uuid_partition() { - let fixture = get_test_fixture(); - let rest_catalog = fixture.rest_catalog().await; - - let schema = Schema::builder() - .with_schema_id(1) - .with_fields(vec![ - NestedField::required(1, "uuid", Type::Primitive(PrimitiveType::Uuid)).into(), - ]) - .build() - .unwrap(); - - let table_creation = TableCreation::builder() - .name("t1".to_string()) - .partition_spec( - UnboundPartitionSpec::builder() - .with_spec_id(0) - .add_partition_fields([UnboundPartitionField::builder() - .source_id(1) - .field_id(1) - .name("uuid".to_string()) - .transform(Transform::Identity) - .build()]) - .unwrap() - .build(), - ) - .schema(schema.clone()) - .build(); - - let ns = random_ns().await; - let table = rest_catalog - .create_table(ns.name(), table_creation) - .await - .unwrap(); - - // Create the writer and write the data - let arrow_schema: Arc = Arc::new( - table - .metadata() - .current_schema() - .as_ref() - .try_into() - .unwrap(), - ); - let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap(); - let file_name_generator = DefaultFileNameGenerator::new( - "test".to_string(), - None, - iceberg::spec::DataFileFormat::Parquet, - ); - let parquet_writer_builder = ParquetWriterBuilder::new( - WriterProperties::default(), - table.metadata().current_schema().clone(), - ); - let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( - parquet_writer_builder, - table.file_io().clone(), - location_generator.clone(), - file_name_generator.clone(), - ); - let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); - - let table_metadata = table.metadata(); - let spec = table_metadata.default_partition_spec().as_ref().clone(); - let table_schema = table_metadata.current_schema().clone(); - - let uuid_value = Uuid::from_u128(0xa1a2a3a4b1b2c1c2d1d2d3d4d5d6d7d8u128); - let uuid_key = Some(Literal::uuid(uuid_value)); - let data = Struct::from_iter([uuid_key]); - let partition_key = PartitionKey::new(spec, table_schema.clone(), data); - - let mut data_file_writer = data_file_writer_builder - .build(Some(partition_key)) - .await - .unwrap(); - let col = FixedSizeBinaryArray::try_from_iter(std::iter::once(uuid_value.as_bytes().to_vec())) - .unwrap(); - let batch = - RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(col) as ArrayRef]).unwrap(); - data_file_writer.write(batch.clone()).await.unwrap(); - let data_file = data_file_writer.close().await.unwrap(); - - let tx = Transaction::new(&table); - let append_action = tx.fast_append().add_data_files(data_file.clone()); - let tx = append_action.apply(tx).unwrap(); - - let table = tx - .commit(&rest_catalog) - .await - .expect("The first commit should not fail."); - - // check results - let batch_stream = table - .scan() - .select_all() - .build() - .unwrap() - .to_arrow() - .await - .unwrap(); - let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); - assert_eq!(batches.len(), 1); - assert_eq!(batches[0], batch); -} 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(()) +}