From 01ba5722a7c968cf196a26d6cbad2d66721a7963 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Sat, 9 May 2026 08:39:43 +0800 Subject: [PATCH 1/5] fix(spec): validate schema types by format version Reject schema types that require a newer table format and update the DataFusion catalog registration path to request v3 only when the converted schema needs it. This keeps ordinary CREATE TABLE defaults on v2 while allowing timestamp_ns schemas to pass validation. Co-authored-by: Codex --- crates/iceberg/src/spec/datatypes.rs | 24 +++ crates/iceberg/src/spec/schema/mod.rs | 50 +++++- crates/iceberg/src/spec/table_metadata.rs | 146 +++++++++++++++++- .../src/spec/table_metadata_builder.rs | 2 + crates/integrations/datafusion/src/schema.rs | 50 +++++- 5 files changed, 261 insertions(+), 11 deletions(-) diff --git a/crates/iceberg/src/spec/datatypes.rs b/crates/iceberg/src/spec/datatypes.rs index 3a112e9f5c..dad1a04adc 100644 --- a/crates/iceberg/src/spec/datatypes.rs +++ b/crates/iceberg/src/spec/datatypes.rs @@ -29,6 +29,7 @@ use serde::de::{Error, IntoDeserializer}; use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; use serde_json::Value as JsonValue; +use super::FormatVersion; use super::values::Literal; use crate::ensure_data_valid; use crate::error::Result; @@ -250,6 +251,29 @@ pub enum PrimitiveType { } impl PrimitiveType { + /// Returns the minimum table format version required by this primitive type. + /// + /// Returns [`None`] if the type is supported in all format versions. + pub const fn min_format_version(&self) -> Option { + match self { + Self::TimestampNs | Self::TimestamptzNs => Some(FormatVersion::V3), + Self::Boolean + | Self::Int + | Self::Long + | Self::Float + | Self::Double + | Self::Decimal { .. } + | Self::Date + | Self::Time + | Self::Timestamp + | Self::Timestamptz + | Self::String + | Self::Uuid + | Self::Fixed(_) + | Self::Binary => None, + } + } + /// Check whether literal is compatible with the type. pub fn compatible(&self, literal: &PrimitiveLiteral) -> bool { matches!( diff --git a/crates/iceberg/src/spec/schema/mod.rs b/crates/iceberg/src/spec/schema/mod.rs index 9109990e19..101be86883 100644 --- a/crates/iceberg/src/spec/schema/mod.rs +++ b/crates/iceberg/src/spec/schema/mod.rs @@ -17,7 +17,7 @@ //! This module defines schema in iceberg. -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt::{Display, Formatter}; use std::sync::Arc; @@ -36,7 +36,7 @@ use self::_serde::SchemaEnum; use self::id_reassigner::ReassignFieldIds; use self::index::{IndexByName, index_by_id, index_parents}; pub use self::prune_columns::prune_columns; -use super::NestedField; +use super::{FormatVersion, NestedField}; use crate::error::Result; use crate::expr::accessor::StructAccessor; use crate::spec::datatypes::{ @@ -421,6 +421,52 @@ impl Schema { pub fn field_id_to_fields(&self) -> &HashMap { &self.id_to_field } + + /// Returns the minimum table format version required by any type in this schema. + /// + /// Returns [`None`] when the schema contains no type with a specific minimum + /// table format version requirement. + pub fn min_format_version(&self) -> Option { + self.id_to_field + .values() + .filter_map(|field| field.field_type.as_primitive_type()) + .filter_map(PrimitiveType::min_format_version) + .max() + } + + /// Validates that all schema types are supported by the table format version. + pub fn validate_compatible_with_format_version( + &self, + format_version: FormatVersion, + ) -> Result<()> { + let problems = self + .id_to_field + .values() + .filter_map(|field| { + let field_type = field.field_type.as_ref(); + let primitive = field_type.as_primitive_type()?; + let min_format_version = primitive.min_format_version()?; + (format_version < min_format_version).then(|| { + let column_name = self.name_by_field_id(field.id).unwrap_or(field.name.as_str()); + ( + field.id, + format!( + "Invalid type for {column_name}: {field_type} is not supported until {min_format_version}" + ), + ) + }) + }) + .collect::>(); + + if !problems.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + problems.into_values().collect::>().join("; "), + )); + } + + Ok(()) + } } impl Display for Schema { diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs index 607fd98350..c4198d78ed 100644 --- a/crates/iceberg/src/spec/table_metadata.rs +++ b/crates/iceberg/src/spec/table_metadata.rs @@ -527,6 +527,7 @@ impl TableMetadata { /// should return normalized `TableMetadata`. pub(super) fn try_normalize(&mut self) -> Result<&mut Self> { self.validate_current_schema()?; + self.validate_schemas_compatible_with_format_version()?; self.normalize_current_snapshot()?; self.construct_refs(); self.validate_refs()?; @@ -540,6 +541,19 @@ impl TableMetadata { Ok(self) } + fn validate_schemas_compatible_with_format_version(&self) -> Result<()> { + let mut schema_ids = self.schemas.keys().copied().collect::>(); + schema_ids.sort_unstable(); + + for schema_id in schema_ids { + if let Some(schema) = self.schemas.get(&schema_id) { + schema.validate_compatible_with_format_version(self.format_version)?; + } + } + + Ok(()) + } + /// If the default partition spec is not present in specs, add it fn try_normalize_partition_spec(&mut self) -> Result<()> { if self @@ -1616,10 +1630,10 @@ mod tests { use crate::io::FileIO; use crate::spec::table_metadata::TableMetadata; use crate::spec::{ - BlobMetadata, EncryptedKey, INITIAL_ROW_ID, Literal, NestedField, NullOrder, Operation, - PartitionSpec, PartitionStatisticsFile, PrimitiveLiteral, PrimitiveType, Schema, Snapshot, - SnapshotReference, SnapshotRetention, SortDirection, SortField, SortOrder, StatisticsFile, - Summary, TableProperties, Transform, Type, UnboundPartitionField, + BlobMetadata, EncryptedKey, INITIAL_ROW_ID, ListType, Literal, NestedField, NullOrder, + Operation, PartitionSpec, PartitionStatisticsFile, PrimitiveLiteral, PrimitiveType, Schema, + Snapshot, SnapshotReference, SnapshotRetention, SortDirection, SortField, SortOrder, + StatisticsFile, Summary, TableProperties, Transform, Type, UnboundPartitionField, }; use crate::{ErrorKind, TableCreation}; @@ -3426,6 +3440,23 @@ mod tests { ) } + #[test] + fn test_table_metadata_v2_rejects_schema_requiring_v3() { + let metadata = + fs::read_to_string("testdata/table_metadata/TableMetadataV2Valid.json").unwrap(); + let mut metadata: serde_json::Value = serde_json::from_str(&metadata).unwrap(); + metadata["schemas"][1]["fields"][2]["type"] = + serde_json::Value::String("timestamp_ns".to_string()); + + let desered: Result = serde_json::from_value(metadata); + + let error_message = desered.unwrap_err().to_string(); + assert!( + error_message.contains("Invalid type for z: timestamp_ns is not supported until v3"), + "expected serde validation to reject v3 type in v2 metadata, got: {error_message}" + ); + } + #[test] fn test_table_metadata_v2_missing_sort_order() { let metadata = @@ -3541,6 +3572,28 @@ mod tests { ) } + fn schema_with_primitive_field(field_type: PrimitiveType) -> Schema { + Schema::builder() + .with_fields(vec![ + NestedField::required(1, "ts", Type::Primitive(field_type)).into(), + ]) + .build() + .unwrap() + } + + fn table_creation_with_format_version( + schema: Schema, + format_version: FormatVersion, + ) -> TableCreation { + TableCreation::builder() + .location("s3://db/table".to_string()) + .name("table".to_string()) + .properties(HashMap::new()) + .schema(schema) + .format_version(format_version) + .build() + } + #[test] fn test_table_metadata_builder_from_table_creation() { let table_creation = TableCreation::builder() @@ -3591,6 +3644,91 @@ mod tests { ); } + #[test] + fn test_table_metadata_builder_rejects_v1_v2_nanosecond_timestamp_tables() { + for (format_version, primitive_type) in [ + (FormatVersion::V1, PrimitiveType::TimestampNs), + (FormatVersion::V1, PrimitiveType::TimestamptzNs), + (FormatVersion::V2, PrimitiveType::TimestampNs), + (FormatVersion::V2, PrimitiveType::TimestamptzNs), + ] { + let table_creation = table_creation_with_format_version( + schema_with_primitive_field(primitive_type), + format_version, + ); + + let err = TableMetadataBuilder::from_table_creation(table_creation).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message().contains("Invalid type for ts:"), + "expected error message to name the invalid column, got {}", + err.message() + ); + assert!( + err.message().contains("is not supported until v3"), + "expected error message to explain v3 requirement, got {}", + err.message() + ); + } + } + + #[test] + fn test_table_metadata_builder_rejects_v2_list_element_requiring_v3() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required( + 1, + "ts_values", + Type::List(ListType::new( + NestedField::list_element( + 2, + Type::Primitive(PrimitiveType::TimestampNs), + false, + ) + .into(), + )), + ) + .into(), + ]) + .build() + .unwrap(); + let table_creation = table_creation_with_format_version(schema, FormatVersion::V2); + + let err = TableMetadataBuilder::from_table_creation(table_creation).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message().contains( + "Invalid type for ts_values.element: timestamp_ns is not supported until v3" + ), + "expected error message to explain nested v3 requirement with column name, got {}", + err.message() + ); + } + + #[test] + fn test_table_metadata_builder_allows_v3_nanosecond_timestamp_tables() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "ts_ns", Type::Primitive(PrimitiveType::TimestampNs)) + .into(), + NestedField::required(2, "tstz_ns", Type::Primitive(PrimitiveType::TimestamptzNs)) + .into(), + ]) + .build() + .unwrap(); + let table_creation = table_creation_with_format_version(schema, FormatVersion::V3); + + let table_metadata = TableMetadataBuilder::from_table_creation(table_creation) + .unwrap() + .build() + .unwrap() + .metadata; + + assert_eq!(table_metadata.format_version, FormatVersion::V3); + } + #[tokio::test] async fn test_table_metadata_read_write() { // Create a temporary directory for our test diff --git a/crates/iceberg/src/spec/table_metadata_builder.rs b/crates/iceberg/src/spec/table_metadata_builder.rs index 3191d6c13c..64647eb549 100644 --- a/crates/iceberg/src/spec/table_metadata_builder.rs +++ b/crates/iceberg/src/spec/table_metadata_builder.rs @@ -638,6 +638,8 @@ impl TableMetadataBuilder { /// Important: Use this method with caution. The builder does not check /// if the added schema is compatible with the current schema. pub fn add_schema(mut self, schema: Schema) -> Result { + schema.validate_compatible_with_format_version(self.metadata.format_version)?; + // Validate that new schema fields don't conflict with existing partition field names self.validate_schema_field_names(&schema)?; diff --git a/crates/integrations/datafusion/src/schema.rs b/crates/integrations/datafusion/src/schema.rs index 76cf599062..53f3a1eab1 100644 --- a/crates/integrations/datafusion/src/schema.rs +++ b/crates/integrations/datafusion/src/schema.rs @@ -162,12 +162,20 @@ impl SchemaProvider for IcebergSchemaProvider { let df_schema = table.schema(); let iceberg_schema = arrow_schema_to_schema_auto_assign_ids(df_schema.as_ref()) .map_err(to_datafusion_error)?; + let format_version = iceberg_schema.min_format_version(); // Create the table in the Iceberg catalog - let table_creation = TableCreation::builder() - .name(name.clone()) - .schema(iceberg_schema) - .build(); + let table_creation = match format_version { + Some(format_version) => TableCreation::builder() + .name(name.clone()) + .format_version(format_version) + .schema(iceberg_schema) + .build(), + None => TableCreation::builder() + .name(name.clone()) + .schema(iceberg_schema) + .build(), + }; let catalog = self.catalog.clone(); let namespace = self.namespace.clone(); @@ -288,10 +296,11 @@ mod tests { use std::sync::Arc; use datafusion::arrow::array::{Int32Array, StringArray}; - use datafusion::arrow::datatypes::{DataType, Field, Schema as ArrowSchema}; + use datafusion::arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit}; use datafusion::arrow::record_batch::RecordBatch; use datafusion::datasource::MemTable; use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; + use iceberg::spec::FormatVersion; use iceberg::{Catalog, CatalogBuilder, NamespaceIdent}; use tempfile::TempDir; @@ -375,6 +384,37 @@ mod tests { assert!(schema_provider.table_exist("empty_table")); } + #[tokio::test] + async fn test_register_timestamp_ns_table_uses_v3() { + let (schema_provider, _temp_dir) = create_test_schema_provider().await; + + let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( + "ts", + DataType::Timestamp(TimeUnit::Nanosecond, None), + true, + )])); + + let empty_batch = RecordBatch::new_empty(arrow_schema.clone()); + let mem_table = MemTable::try_new(arrow_schema, vec![vec![empty_batch]]).unwrap(); + + let result = + schema_provider.register_table("timestamp_ns_table".to_string(), Arc::new(mem_table)); + + assert!(result.is_ok(), "Expected success, got: {result:?}"); + + let table_ident = TableIdent::new( + schema_provider.namespace.clone(), + "timestamp_ns_table".to_string(), + ); + let table = schema_provider + .catalog + .load_table(&table_ident) + .await + .unwrap(); + + assert_eq!(FormatVersion::V3, table.metadata().format_version()); + } + #[tokio::test] async fn test_register_duplicate_table_fails() { let (schema_provider, _temp_dir) = create_test_schema_provider().await; From 917b2325747c048ebd5ea703f8f19a75a7d2b6a7 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Sat, 30 May 2026 23:35:43 +0800 Subject: [PATCH 2/5] fix(spec): clarify primitive format compatibility Move the table-format comparison behind PrimitiveType so schema validation asks the primitive for the required format version only when the requested format does not support it. Co-authored-by: Codex --- crates/iceberg/public-api.txt | 1 + crates/iceberg/src/spec/datatypes.rs | 12 +++++++++++- crates/iceberg/src/spec/schema/mod.rs | 21 ++++++++++----------- 3 files changed, 22 insertions(+), 12 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 41d4d7691b..11ef635e0f 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -2313,6 +2313,7 @@ pub fn iceberg::spec::Schema::field_id_to_name_map(&self) -> &std::collections:: pub fn iceberg::spec::Schema::highest_field_id(&self) -> i32 pub fn iceberg::spec::Schema::identifier_field_ids(&self) -> impl core::iter::traits::exact_size::ExactSizeIterator + '_ pub fn iceberg::spec::Schema::into_builder(self) -> iceberg::spec::SchemaBuilder +pub fn iceberg::spec::Schema::min_format_version(&self) -> core::option::Option pub fn iceberg::spec::Schema::name_by_field_id(&self, field_id: i32) -> core::option::Option<&str> pub fn iceberg::spec::Schema::schema_id(&self) -> iceberg::spec::SchemaId impl core::clone::Clone for iceberg::spec::Schema diff --git a/crates/iceberg/src/spec/datatypes.rs b/crates/iceberg/src/spec/datatypes.rs index dad1a04adc..f237cf4873 100644 --- a/crates/iceberg/src/spec/datatypes.rs +++ b/crates/iceberg/src/spec/datatypes.rs @@ -254,7 +254,7 @@ impl PrimitiveType { /// Returns the minimum table format version required by this primitive type. /// /// Returns [`None`] if the type is supported in all format versions. - pub const fn min_format_version(&self) -> Option { + pub(super) const fn min_format_version(&self) -> Option { match self { Self::TimestampNs | Self::TimestamptzNs => Some(FormatVersion::V3), Self::Boolean @@ -274,6 +274,16 @@ impl PrimitiveType { } } + /// Returns the required table format version when this primitive type is + /// not supported by `format_version`. + pub(super) fn required_format_version_if_unsupported( + &self, + format_version: FormatVersion, + ) -> Option { + self.min_format_version() + .filter(|min_format_version| format_version < *min_format_version) + } + /// Check whether literal is compatible with the type. pub fn compatible(&self, literal: &PrimitiveLiteral) -> bool { matches!( diff --git a/crates/iceberg/src/spec/schema/mod.rs b/crates/iceberg/src/spec/schema/mod.rs index 101be86883..d3c510e6b7 100644 --- a/crates/iceberg/src/spec/schema/mod.rs +++ b/crates/iceberg/src/spec/schema/mod.rs @@ -435,7 +435,7 @@ impl Schema { } /// Validates that all schema types are supported by the table format version. - pub fn validate_compatible_with_format_version( + pub(super) fn validate_compatible_with_format_version( &self, format_version: FormatVersion, ) -> Result<()> { @@ -445,16 +445,15 @@ impl Schema { .filter_map(|field| { let field_type = field.field_type.as_ref(); let primitive = field_type.as_primitive_type()?; - let min_format_version = primitive.min_format_version()?; - (format_version < min_format_version).then(|| { - let column_name = self.name_by_field_id(field.id).unwrap_or(field.name.as_str()); - ( - field.id, - format!( - "Invalid type for {column_name}: {field_type} is not supported until {min_format_version}" - ), - ) - }) + let required_format_version = + primitive.required_format_version_if_unsupported(format_version)?; + let column_name = self.name_by_field_id(field.id).unwrap_or(field.name.as_str()); + Some(( + field.id, + format!( + "Invalid type for {column_name}: {field_type} is not supported until {required_format_version}" + ), + )) }) .collect::>(); From 019da08278fe656ea347d734138766636ef38f5a Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Thu, 7 May 2026 22:20:56 +0800 Subject: [PATCH 3/5] feat(spec): add unknown datatype support Add Iceberg unknown type parsing and conversions across schema, Arrow, Avro, HMS, and Glue paths. Reject non-null defaults for unknown fields and make set predicate formatting deterministic for stable tests. Co-authored-by: Codex --- crates/catalog/glue/src/schema.rs | 6 + crates/catalog/hms/src/schema.rs | 6 + crates/iceberg/src/arrow/reader/projection.rs | 2 + crates/iceberg/src/arrow/schema.rs | 13 ++ crates/iceberg/src/arrow/value.rs | 30 ++++- crates/iceberg/src/avro/schema.rs | 61 ++++++--- crates/iceberg/src/expr/predicate.rs | 19 ++- .../src/expr/visitors/strict_projection.rs | 24 ++-- crates/iceberg/src/spec/datatypes.rs | 51 +++++--- crates/iceberg/src/spec/schema/mod.rs | 116 ++++++++++++++++++ crates/iceberg/src/spec/values/datum.rs | 6 + crates/iceberg/src/transform/bucket.rs | 6 +- crates/iceberg/src/transform/temporal.rs | 38 +++--- crates/iceberg/src/transform/truncate.rs | 10 +- 14 files changed, 303 insertions(+), 85 deletions(-) diff --git a/crates/catalog/glue/src/schema.rs b/crates/catalog/glue/src/schema.rs index 864320dae4..86ad86ff96 100644 --- a/crates/catalog/glue/src/schema.rs +++ b/crates/catalog/glue/src/schema.rs @@ -157,6 +157,12 @@ impl SchemaVisitor for GlueSchemaBuilder { fn primitive(&mut self, p: &iceberg::spec::PrimitiveType) -> iceberg::Result { let glue_type = match p { + PrimitiveType::Unknown => { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Conversion from {p:?} is not supported"), + )); + } PrimitiveType::Boolean => "boolean".to_string(), PrimitiveType::Int => "int".to_string(), PrimitiveType::Long => "bigint".to_string(), diff --git a/crates/catalog/hms/src/schema.rs b/crates/catalog/hms/src/schema.rs index c23b48719d..6cac2d1b22 100644 --- a/crates/catalog/hms/src/schema.rs +++ b/crates/catalog/hms/src/schema.rs @@ -114,6 +114,12 @@ impl SchemaVisitor for HiveSchemaBuilder { fn primitive(&mut self, p: &iceberg::spec::PrimitiveType) -> iceberg::Result { let hive_type = match p { + PrimitiveType::Unknown => { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Conversion from {p:?} is not supported"), + )); + } PrimitiveType::Boolean => "boolean".to_string(), PrimitiveType::Int => "int".to_string(), PrimitiveType::Long => "bigint".to_string(), diff --git a/crates/iceberg/src/arrow/reader/projection.rs b/crates/iceberg/src/arrow/reader/projection.rs index aad3123f0a..2bbb76bb8a 100644 --- a/crates/iceberg/src/arrow/reader/projection.rs +++ b/crates/iceberg/src/arrow/reader/projection.rs @@ -66,6 +66,7 @@ impl ArrowReader { /// Nested types (struct/list/map) are flattened in Parquet's columnar format. fn include_leaf_field_id(field: &NestedField, field_ids: &mut Vec) { match field.field_type.as_ref() { + Type::Primitive(PrimitiveType::Unknown) => {} Type::Primitive(_) => { field_ids.push(field.id); } @@ -99,6 +100,7 @@ impl ArrowReader { (Some(lhs), Some(rhs)) if lhs == rhs => true, (Some(PrimitiveType::Int), Some(PrimitiveType::Long)) => true, (Some(PrimitiveType::Float), Some(PrimitiveType::Double)) => true, + (Some(PrimitiveType::Unknown), Some(_)) => true, ( Some(PrimitiveType::Decimal { precision: file_precision, diff --git a/crates/iceberg/src/arrow/schema.rs b/crates/iceberg/src/arrow/schema.rs index e6649d926d..57e1a13e21 100644 --- a/crates/iceberg/src/arrow/schema.rs +++ b/crates/iceberg/src/arrow/schema.rs @@ -120,6 +120,7 @@ fn visit_type(r#type: &DataType, visitor: &mut V) -> Resu | DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View + | DataType::Null | DataType::Binary | DataType::LargeBinary | DataType::BinaryView @@ -428,6 +429,7 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter { fn primitive(&mut self, p: &DataType) -> Result { match p { + DataType::Null => Ok(Type::Primitive(PrimitiveType::Unknown)), DataType::Boolean => Ok(Type::Primitive(PrimitiveType::Boolean)), DataType::Int8 | DataType::Int16 | DataType::Int32 => { Ok(Type::Primitive(PrimitiveType::Int)) @@ -613,6 +615,9 @@ impl SchemaVisitor for ToArrowSchemaConverter { p: &crate::spec::PrimitiveType, ) -> crate::Result { match p { + crate::spec::PrimitiveType::Unknown => { + Ok(ArrowSchemaOrFieldOrType::Type(DataType::Null)) + } crate::spec::PrimitiveType::Boolean => { Ok(ArrowSchemaOrFieldOrType::Type(DataType::Boolean)) } @@ -1116,6 +1121,7 @@ pub fn datum_to_arrow_type_with_ree(datum: &Datum) -> DataType { // Match on the PrimitiveType from the Datum to determine the Arrow type match datum.data_type() { + PrimitiveType::Unknown => make_ree(DataType::Null), PrimitiveType::Boolean => make_ree(DataType::Boolean), PrimitiveType::Int => make_ree(DataType::Int32), PrimitiveType::Long => make_ree(DataType::Int64), @@ -1915,6 +1921,13 @@ mod tests { assert_eq!(iceberg_type, arrow_type_to_type(&arrow_type).unwrap()); } + { + let arrow_type = DataType::Null; + let iceberg_type = Type::Primitive(PrimitiveType::Unknown); + assert_eq!(arrow_type, type_to_arrow_type(&iceberg_type).unwrap()); + assert_eq!(iceberg_type, arrow_type_to_type(&arrow_type).unwrap()); + } + // test struct type { // no metadata will cause error diff --git a/crates/iceberg/src/arrow/value.rs b/crates/iceberg/src/arrow/value.rs index d07233c420..39d646c241 100644 --- a/crates/iceberg/src/arrow/value.rs +++ b/crates/iceberg/src/arrow/value.rs @@ -206,6 +206,7 @@ impl SchemaWithPartnerVisitor for ArrowArrayToIcebergStructConverter { fn primitive(&mut self, p: &PrimitiveType, partner: &ArrayRef) -> Result>> { match p { + PrimitiveType::Unknown => Ok(vec![None; partner.len()]), PrimitiveType::Boolean => { let array = partner .as_any() @@ -629,6 +630,7 @@ pub(crate) fn create_primitive_array_single_element( prim_lit: &Option, ) -> Result { match (data_type, prim_lit) { + (DataType::Null, None) => Ok(Arc::new(arrow_array::NullArray::new(1))), (DataType::Boolean, Some(PrimitiveLiteral::Boolean(v))) => { Ok(Arc::new(BooleanArray::from(vec![*v]))) } @@ -965,7 +967,7 @@ pub(crate) fn create_primitive_array_repeated( Some(NullBuffer::new_null(num_rows)), )) } - (DataType::Null, _) => Arc::new(arrow_array::NullArray::new(num_rows)), + (DataType::Null, None) => Arc::new(arrow_array::NullArray::new(num_rows)), (dt, _) => { return Err(Error::new( ErrorKind::Unexpected, @@ -1847,6 +1849,32 @@ mod test { } } + #[test] + fn test_create_null_array_rejects_non_null_literal() { + let single = create_primitive_array_single_element( + &DataType::Null, + &Some(PrimitiveLiteral::Boolean(true)), + ); + assert!(single.is_err()); + + let repeated = create_primitive_array_repeated( + &DataType::Null, + &Some(PrimitiveLiteral::Boolean(true)), + 3, + ); + assert!(repeated.is_err()); + + let single_null = create_primitive_array_single_element(&DataType::Null, &None) + .expect("Failed to create single null array"); + assert_eq!(single_null.data_type(), &DataType::Null); + assert_eq!(single_null.len(), 1); + + let repeated_null = create_primitive_array_repeated(&DataType::Null, &None, 3) + .expect("Failed to create repeated null array"); + assert_eq!(repeated_null.data_type(), &DataType::Null); + assert_eq!(repeated_null.len(), 3); + } + #[test] fn test_create_decimal_array_repeated_respects_precision() { // Ensure repeated arrays also respect target precision, not Arrow's default. diff --git a/crates/iceberg/src/avro/schema.rs b/crates/iceberg/src/avro/schema.rs index fdbc680977..e2641a6b41 100644 --- a/crates/iceberg/src/avro/schema.rs +++ b/crates/iceberg/src/avro/schema.rs @@ -74,7 +74,7 @@ impl SchemaVisitor for SchemaToAvroSchema { record.name = Name::from(format!("r{}", field.id).as_str()); } - if !field.required { + if !field.required && !matches!(field_schema, AvroSchema::Null) { field_schema = avro_optional(field_schema)?; } @@ -126,7 +126,7 @@ impl SchemaVisitor for SchemaToAvroSchema { record.name = Name::from(format!("r{}", list.element_field.id).as_str()); } - if !list.element_field.required { + if !list.element_field.required && !matches!(field_schema, AvroSchema::Null) { field_schema = avro_optional(field_schema)?; } @@ -147,7 +147,7 @@ impl SchemaVisitor for SchemaToAvroSchema { ) -> Result { let key_field_schema = key_value.unwrap_left(); let mut value_field_schema = value.unwrap_left(); - if !map.value_field.required { + if !map.value_field.required && !matches!(value_field_schema, AvroSchema::Null) { value_field_schema = avro_optional(value_field_schema)?; } @@ -222,6 +222,7 @@ impl SchemaVisitor for SchemaToAvroSchema { fn primitive(&mut self, p: &PrimitiveType) -> Result { let avro_schema = match p { + PrimitiveType::Unknown => AvroSchema::Null, PrimitiveType::Boolean => AvroSchema::Boolean, PrimitiveType::Int => AvroSchema::Int, PrimitiveType::Long => AvroSchema::Long, @@ -304,6 +305,10 @@ pub(crate) fn avro_decimal_schema(precision: usize, scale: usize) -> Result Result { + if matches!(avro_schema, AvroSchema::Null) { + return Ok(AvroSchema::Null); + } + Ok(AvroSchema::Union(UnionSchema::new(vec![ AvroSchema::Null, avro_schema, @@ -440,10 +445,11 @@ impl AvroSchemaVisitor for AvroSchemaToSchema { let field_id = Self::get_element_id_from_attributes(&avro_field.custom_attributes, FIELD_ID_PROP)?; - let optional = is_avro_optional(&avro_field.schema); + let optional = is_avro_optional(&avro_field.schema) + || matches!(&avro_field.schema, AvroSchema::Null); - let mut field = - NestedField::new(field_id, &avro_field.name, field_type.unwrap(), !optional); + let field_type = field_type.unwrap_or(Type::Primitive(PrimitiveType::Unknown)); + let mut field = NestedField::new(field_id, &avro_field.name, field_type, !optional); if let Some(doc) = &avro_field.doc { field = field.with_doc(doc); @@ -475,7 +481,9 @@ impl AvroSchemaVisitor for AvroSchemaToSchema { } if options.len() == 1 { - Ok(Some(options.remove(0).unwrap())) + Ok(options + .remove(0) + .or(Some(Type::Primitive(PrimitiveType::Unknown)))) } else { Ok(Some(options.remove(1).unwrap())) } @@ -483,10 +491,11 @@ impl AvroSchemaVisitor for AvroSchemaToSchema { fn array(&mut self, array: &ArraySchema, item: Option) -> Result { let element_field_id = Self::get_element_id_from_attributes(&array.attributes, ELEMENT_ID)?; + let item = item.unwrap_or(Type::Primitive(PrimitiveType::Unknown)); let element_field = NestedField::list_element( element_field_id, - item.unwrap(), - !is_avro_optional(&array.items), + item, + !is_avro_optional(&array.items) && !matches!(array.items.as_ref(), AvroSchema::Null), ) .into(); Ok(Some(Type::List(ListType { element_field }))) @@ -497,10 +506,11 @@ impl AvroSchemaVisitor for AvroSchemaToSchema { let key_field = NestedField::map_key_element(key_field_id, Type::Primitive(PrimitiveType::String)); let value_field_id = Self::get_element_id_from_attributes(&map.attributes, VALUE_ID)?; + let value = value.unwrap_or(Type::Primitive(PrimitiveType::Unknown)); let value_field = NestedField::map_value_element( value_field_id, - value.unwrap(), - !is_avro_optional(&map.types), + value, + !is_avro_optional(&map.types) && !matches!(map.types.as_ref(), AvroSchema::Null), ); Ok(Some(Type::Map(MapType { key_field: key_field.into(), @@ -550,12 +560,7 @@ impl AvroSchemaVisitor for AvroSchemaToSchema { "Can't convert avro map schema, missing key schema.", ) })?; - let value = value.ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - "Can't convert avro map schema, missing value schema.", - ) - })?; + let value = value.unwrap_or(Type::Primitive(PrimitiveType::Unknown)); let key_id = Self::get_element_id_from_attributes( &array.fields[0].custom_attributes, FIELD_ID_PROP, @@ -568,7 +573,8 @@ impl AvroSchemaVisitor for AvroSchemaToSchema { let value_field = NestedField::map_value_element( value_id, value, - !is_avro_optional(&array.fields[1].schema), + !is_avro_optional(&array.fields[1].schema) + && !matches!(&array.fields[1].schema, AvroSchema::Null), ); Ok(Some(Type::Map(MapType { key_field: key_field.into(), @@ -650,6 +656,25 @@ mod tests { assert_eq!(iceberg_schema, converted_avro_converted_iceberg_schema); } + #[test] + fn test_unknown_type_schema_conversion() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "empty", PrimitiveType::Unknown.into()).into(), + ]) + .build() + .unwrap(); + + let avro_schema = schema_to_avro_schema("table", &schema).unwrap(); + let AvroSchema::Record(record) = &avro_schema else { + panic!("expected avro record schema"); + }; + assert!(matches!(record.fields[0].schema, AvroSchema::Null)); + assert_eq!(record.fields[0].default, Some(Value::Null)); + + assert_eq!(schema, avro_schema_to_schema(&avro_schema).unwrap()); + } + #[test] fn test_manifest_file_v1_schema() { let fields = vec![ diff --git a/crates/iceberg/src/expr/predicate.rs b/crates/iceberg/src/expr/predicate.rs index 5a7330c8ca..d99ed36674 100644 --- a/crates/iceberg/src/expr/predicate.rs +++ b/crates/iceberg/src/expr/predicate.rs @@ -310,7 +310,12 @@ impl Bind for SetExpression { impl Display for SetExpression { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - let mut literal_strs = self.literals.iter().map(|l| format!("{l}")); + let mut literals = self.literals.iter().collect_vec(); + literals.sort_by(|left, right| { + left.partial_cmp(right) + .unwrap_or_else(|| left.to_string().cmp(&right.to_string())) + }); + let mut literal_strs = literals.into_iter().map(ToString::to_string); write!(f, "{} {} ({})", self.term, self.op, literal_strs.join(", ")) } @@ -1363,7 +1368,7 @@ mod tests { let schema = table_schema_simple(); let expr = Reference::new("bar").is_in([Datum::int(10), Datum::int(20)]); let bound_expr = expr.bind(schema, true).unwrap(); - assert_eq!(&format!("{bound_expr}"), "bar IN (20, 10)"); + assert_eq!(&format!("{bound_expr}"), "bar IN (10, 20)"); test_bound_predicate_serialize_diserialize(bound_expr); } @@ -1398,7 +1403,7 @@ mod tests { let schema = table_schema_simple(); let expr = Reference::new("bar").is_not_in([Datum::int(10), Datum::int(20)]); let bound_expr = expr.bind(schema, true).unwrap(); - assert_eq!(&format!("{bound_expr}"), "bar NOT IN (20, 10)"); + assert_eq!(&format!("{bound_expr}"), "bar NOT IN (10, 20)"); test_bound_predicate_serialize_diserialize(bound_expr); } @@ -1571,13 +1576,7 @@ mod tests { let expected_bound = expected_predicate.bind(schema, true).unwrap(); assert_eq!(result, expected_bound); - // Note: HashSet order may vary, so we check that it contains the expected format - let result_str = format!("{result}"); - assert!( - result_str.contains("bar NOT IN") - && result_str.contains("10") - && result_str.contains("20") - ); + assert_eq!(&format!("{result}"), "bar NOT IN (10, 20)"); } #[test] diff --git a/crates/iceberg/src/expr/visitors/strict_projection.rs b/crates/iceberg/src/expr/visitors/strict_projection.rs index ebc6212c76..0e8f5c18c9 100644 --- a/crates/iceberg/src/expr/visitors/strict_projection.rs +++ b/crates/iceberg/src/expr/visitors/strict_projection.rs @@ -424,7 +424,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "((((pcol1 NOT IN (0, -1)) AND (pcol2 NOT IN (0, -1))) AND (pcol3 NOT IN (0, -1))) AND (pcol4 NOT IN (0, -1))) AND (pcol5 NOT IN (0, -1))".to_string()); + assert_eq!(result.to_string(), "((((pcol1 NOT IN (-1, 0)) AND (pcol2 NOT IN (-1, 0))) AND (pcol3 NOT IN (-1, 0))) AND (pcol4 NOT IN (-1, 0))) AND (pcol5 NOT IN (-1, 0))".to_string()); // test in let predicate = @@ -658,7 +658,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "((((pcol1 NOT IN (575, 564)) AND (pcol2 NOT IN (575, 564))) AND (pcol3 NOT IN (575, 564))) AND (pcol4 NOT IN (575, 564))) AND (pcol5 NOT IN (575, 564))".to_string()); + assert_eq!(result.to_string(), "((((pcol1 NOT IN (564, 575)) AND (pcol2 NOT IN (564, 575))) AND (pcol3 NOT IN (564, 575))) AND (pcol4 NOT IN (564, 575))) AND (pcol5 NOT IN (564, 575))".to_string()); // test in let predicate = Reference::new("col1") @@ -886,7 +886,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "((((pcol1 NOT IN (-1, -12)) AND (pcol2 NOT IN (-1, -12))) AND (pcol3 NOT IN (-1, -12))) AND (pcol4 NOT IN (-1, -12))) AND (pcol5 NOT IN (-1, -12))".to_string()); + assert_eq!(result.to_string(), "((((pcol1 NOT IN (-12, -1)) AND (pcol2 NOT IN (-12, -1))) AND (pcol3 NOT IN (-12, -1))) AND (pcol4 NOT IN (-12, -1))) AND (pcol5 NOT IN (-12, -1))".to_string()); // test in let predicate = Reference::new("col1") @@ -1114,7 +1114,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "((((pcol1 NOT IN (575, 564)) AND (pcol2 NOT IN (575, 564))) AND (pcol3 NOT IN (575, 564))) AND (pcol4 NOT IN (575, 564))) AND (pcol5 NOT IN (575, 564))".to_string()); + assert_eq!(result.to_string(), "((((pcol1 NOT IN (564, 575)) AND (pcol2 NOT IN (564, 575))) AND (pcol3 NOT IN (564, 575))) AND (pcol4 NOT IN (564, 575))) AND (pcol5 NOT IN (564, 575))".to_string()); // test in let predicate = Reference::new("col1") @@ -1724,7 +1724,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "((((pcol1 NOT IN (1969-12-31, 1969-12-30)) AND (pcol2 NOT IN (1969-12-31, 1969-12-30))) AND (pcol3 NOT IN (1969-12-31, 1969-12-30))) AND (pcol4 NOT IN (1969-12-31, 1969-12-30))) AND (pcol5 NOT IN (1969-12-31, 1969-12-30))".to_string()); + assert_eq!(result.to_string(), "((((pcol1 NOT IN (1969-12-30, 1969-12-31)) AND (pcol2 NOT IN (1969-12-30, 1969-12-31))) AND (pcol3 NOT IN (1969-12-30, 1969-12-31))) AND (pcol4 NOT IN (1969-12-30, 1969-12-31))) AND (pcol5 NOT IN (1969-12-30, 1969-12-31))".to_string()); // test in let predicate = Reference::new("col1") @@ -1952,7 +1952,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "((((pcol1 NOT IN (47, 46)) AND (pcol2 NOT IN (47, 46))) AND (pcol3 NOT IN (47, 46))) AND (pcol4 NOT IN (47, 46))) AND (pcol5 NOT IN (47, 46))".to_string()); + assert_eq!(result.to_string(), "((((pcol1 NOT IN (46, 47)) AND (pcol2 NOT IN (46, 47))) AND (pcol3 NOT IN (46, 47))) AND (pcol4 NOT IN (46, 47))) AND (pcol5 NOT IN (46, 47))".to_string()); // test in let predicate = Reference::new("col1") @@ -2384,7 +2384,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "((((pcol1 NOT IN (47, 46)) AND (pcol2 NOT IN (47, 46))) AND (pcol3 NOT IN (47, 46))) AND (pcol4 NOT IN (47, 46))) AND (pcol5 NOT IN (47, 46))".to_string()); + assert_eq!(result.to_string(), "((((pcol1 NOT IN (46, 47)) AND (pcol2 NOT IN (46, 47))) AND (pcol3 NOT IN (46, 47))) AND (pcol4 NOT IN (46, 47))) AND (pcol5 NOT IN (46, 47))".to_string()); // test in let predicate = Reference::new("col1") @@ -2593,7 +2593,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "(((((pcol1 NOT IN (8, 7, 6)) AND (pcol2 NOT IN (8, 7, 6))) AND (pcol3 NOT IN (6, 2))) AND (pcol4 NOT IN (9, 4))) AND (pcol5 NOT IN (4, 6))) AND (pcol6 NOT IN (4, 6))".to_string()); + assert_eq!(result.to_string(), "(((((pcol1 NOT IN (6, 7, 8)) AND (pcol2 NOT IN (6, 7, 8))) AND (pcol3 NOT IN (2, 6))) AND (pcol4 NOT IN (4, 9))) AND (pcol5 NOT IN (4, 6))) AND (pcol6 NOT IN (4, 6))".to_string()); } #[tokio::test] @@ -2690,7 +2690,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "pcol1 IN (101, 100)".to_string()); + assert_eq!(result.to_string(), "pcol1 IN (100, 101)".to_string()); // test not in let predicate = Reference::new("col1") @@ -2698,7 +2698,7 @@ mod tests { .bind(schema.clone(), false) .unwrap(); let result = strict_projection.strict_project(&predicate).unwrap(); - assert_eq!(result.to_string(), "pcol1 NOT IN (101, 100)".to_string()); + assert_eq!(result.to_string(), "pcol1 NOT IN (100, 101)".to_string()); } #[tokio::test] @@ -2880,7 +2880,7 @@ mod tests { let result = strict_projection.strict_project(&predicate).unwrap(); assert_eq!( result.to_string(), - "((pcol1 NOT IN (100, 90)) AND (pcol2 NOT IN (100, 90))) AND (pcol3 NOT IN (10000, 10100, 9900))" + "((pcol1 NOT IN (90, 100)) AND (pcol2 NOT IN (90, 100))) AND (pcol3 NOT IN (9900, 10000, 10100))" .to_string() ); @@ -3103,7 +3103,7 @@ mod tests { let result = strict_projection.strict_project(&predicate).unwrap(); assert_eq!( result.to_string(), - "((pcol1 NOT IN (100, 90)) AND (pcol2 NOT IN (100, 90))) AND (pcol3 NOT IN (9890, 9990, 10090))" + "((pcol1 NOT IN (90, 100)) AND (pcol2 NOT IN (90, 100))) AND (pcol3 NOT IN (9890, 9990, 10090))" .to_string() ); diff --git a/crates/iceberg/src/spec/datatypes.rs b/crates/iceberg/src/spec/datatypes.rs index f237cf4873..3072e76cd2 100644 --- a/crates/iceberg/src/spec/datatypes.rs +++ b/crates/iceberg/src/spec/datatypes.rs @@ -19,7 +19,6 @@ * Data Types */ use std::collections::HashMap; -use std::convert::identity; use std::fmt; use std::ops::Index; use std::sync::{Arc, OnceLock}; @@ -209,6 +208,8 @@ impl From for Type { #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash)] #[serde(rename_all = "lowercase", remote = "Self")] pub enum PrimitiveType { + /// Default / null column type used when a more specific type is not known. + Unknown, /// True or False Boolean, /// 32-bit signed integer @@ -257,7 +258,8 @@ impl PrimitiveType { pub(super) const fn min_format_version(&self) -> Option { match self { Self::TimestampNs | Self::TimestamptzNs => Some(FormatVersion::V3), - Self::Boolean + Self::Unknown + | Self::Boolean | Self::Int | Self::Long | Self::Float @@ -398,6 +400,7 @@ where S: Serializer { impl fmt::Display for PrimitiveType { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match self { + PrimitiveType::Unknown => write!(f, "unknown"), PrimitiveType::Boolean => write!(f, "boolean"), PrimitiveType::Int => write!(f, "int"), PrimitiveType::Long => write!(f, "long"), @@ -561,7 +564,7 @@ impl fmt::Display for StructType { } #[derive(Debug, PartialEq, Serialize, Deserialize, Eq, Clone)] -#[serde(from = "SerdeNestedField", into = "SerdeNestedField")] +#[serde(try_from = "SerdeNestedField", into = "SerdeNestedField")] /// A struct is a tuple of typed values. Each field in the tuple is named and has an integer id that is unique in the table schema. /// Each field can be either optional or required, meaning that values can (or cannot) be null. Fields may be any type. /// Fields may have an optional comment or doc string. Fields can have default values. @@ -598,25 +601,30 @@ struct SerdeNestedField { pub write_default: Option, } -impl From for NestedField { - fn from(value: SerdeNestedField) -> Self { - NestedField { +impl TryFrom for NestedField { + type Error = crate::Error; + + fn try_from(value: SerdeNestedField) -> Result { + let initial_default = value + .initial_default + .map(|x| Literal::try_from_json(x, &value.field_type)) + .transpose()? + .flatten(); + let write_default = value + .write_default + .map(|x| Literal::try_from_json(x, &value.field_type)) + .transpose()? + .flatten(); + + Ok(NestedField { id: value.id, name: value.name, required: value.required, - initial_default: value.initial_default.and_then(|x| { - Literal::try_from_json(x, &value.field_type) - .ok() - .and_then(identity) - }), - write_default: value.write_default.and_then(|x| { - Literal::try_from_json(x, &value.field_type) - .ok() - .and_then(identity) - }), + initial_default, + write_default, field_type: value.field_type, doc: value.doc, - } + }) } } @@ -903,6 +911,7 @@ mod tests { { "type": "struct", "fields": [ + {"id": 17, "name": "unknown_field", "required": false, "type": "unknown"}, {"id": 1, "name": "bool_field", "required": true, "type": "boolean"}, {"id": 2, "name": "int_field", "required": true, "type": "int"}, {"id": 3, "name": "long_field", "required": true, "type": "long"}, @@ -927,6 +936,12 @@ mod tests { record, Type::Struct(StructType { fields: vec![ + NestedField::optional( + 17, + "unknown_field", + Type::Primitive(PrimitiveType::Unknown), + ) + .into(), NestedField::required(1, "bool_field", Type::Primitive(PrimitiveType::Boolean)) .into(), NestedField::required(2, "int_field", Type::Primitive(PrimitiveType::Int)) @@ -1308,6 +1323,8 @@ mod tests { for (ty, literal) in pairs { assert!(ty.compatible(&literal)); } + + assert!(!PrimitiveType::Unknown.compatible(&PrimitiveLiteral::Int(1))); } #[test] diff --git a/crates/iceberg/src/spec/schema/mod.rs b/crates/iceberg/src/spec/schema/mod.rs index d3c510e6b7..2e49e6d15a 100644 --- a/crates/iceberg/src/spec/schema/mod.rs +++ b/crates/iceberg/src/spec/schema/mod.rs @@ -129,6 +129,10 @@ impl SchemaBuilder { /// Builds the schema. pub fn build(self) -> Result { + for field in &self.fields { + Self::validate_unknown_type_field(field)?; + } + let field_id_to_accessor = self.build_accessors(); let r#struct = StructType::new(self.fields); @@ -186,6 +190,38 @@ impl SchemaBuilder { Ok(schema) } + fn validate_unknown_type_field(field: &NestedFieldRef) -> Result<()> { + match field.field_type.as_ref() { + Type::Primitive(PrimitiveType::Unknown) => { + ensure_data_valid!( + !field.required, + "Field {} cannot be required because unknown type must be optional", + field.name + ); + ensure_data_valid!( + field.initial_default.is_none() && field.write_default.is_none(), + "Field {} cannot have non-null defaults because unknown type requires null defaults", + field.name + ); + } + Type::Struct(struct_type) => { + for nested_field in struct_type.fields() { + Self::validate_unknown_type_field(nested_field)?; + } + } + Type::List(list_type) => { + Self::validate_unknown_type_field(&list_type.element_field)?; + } + Type::Map(map_type) => { + Self::validate_unknown_type_field(&map_type.key_field)?; + Self::validate_unknown_type_field(&map_type.value_field)?; + } + Type::Primitive(_) => {} + } + + Ok(()) + } + fn build_accessors(&self) -> HashMap> { let mut map = HashMap::new(); @@ -1275,4 +1311,84 @@ table { .is_err() ); } + + #[test] + fn test_unknown_type_deserialization_rejects_non_null_default() { + let schema_json = serde_json::json!({ + "type": "struct", + "schema-id": 1, + "fields": [ + { + "id": 1, + "name": "empty", + "required": false, + "type": "unknown", + "initial-default": 1 + } + ] + }); + + assert!(serde_json::from_value::(schema_json).is_err()); + } + + #[test] + fn test_unknown_type_deserialization_accepts_null_defaults() { + let schema_json = serde_json::json!({ + "type": "struct", + "schema-id": 1, + "fields": [ + { + "id": 1, + "name": "empty", + "required": false, + "type": "unknown", + "initial-default": null, + "write-default": null + } + ] + }); + + serde_json::from_value::(schema_json).unwrap(); + } + + #[test] + fn test_unknown_type_must_be_optional_with_null_defaults() { + assert!( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::optional(1, "empty", Primitive(PrimitiveType::Unknown)).into() + ]) + .build() + .is_ok() + ); + + let required_error = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "empty", Primitive(PrimitiveType::Unknown)).into(), + ]) + .build() + .unwrap_err(); + assert!( + required_error + .message() + .contains("unknown type must be optional") + ); + + let default_error = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::optional(1, "empty", Primitive(PrimitiveType::Unknown)) + .with_initial_default(Literal::int(1)) + .into(), + ]) + .build() + .unwrap_err(); + assert!( + default_error + .message() + .contains("unknown type requires null defaults") + ); + } } diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index 2040d4634a..3f0ead00a8 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -368,6 +368,12 @@ impl Datum { /// See [this spec](https://iceberg.apache.org/spec/#binary-single-value-serialization) for reference. pub fn try_from_bytes(bytes: &[u8], data_type: PrimitiveType) -> Result { let literal = match data_type { + PrimitiveType::Unknown => { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + "Cannot create datum for unknown type from bytes", + )); + } PrimitiveType::Boolean => { if bytes.len() == 1 && bytes[0] == 0u8 { PrimitiveLiteral::Boolean(false) diff --git a/crates/iceberg/src/transform/bucket.rs b/crates/iceberg/src/transform/bucket.rs index 52fb72f1cf..befdb019f3 100644 --- a/crates/iceberg/src/transform/bucket.rs +++ b/crates/iceberg/src/transform/bucket.rs @@ -519,7 +519,7 @@ mod test { Datum::string(value), Datum::string(another), ]), - Some("name IN (9, 4)"), + Some("name IN (4, 9)"), )?; fixture.assert_projection( @@ -656,7 +656,7 @@ mod test { Datum::long(value), Datum::long(value + 1), ]), - Some("name IN (8, 7, 6)"), + Some("name IN (6, 7, 8)"), )?; fixture.assert_projection( @@ -716,7 +716,7 @@ mod test { Datum::int(value), Datum::int(value + 1), ]), - Some("name IN (8, 7, 6)"), + Some("name IN (6, 7, 8)"), )?; fixture.assert_projection( diff --git a/crates/iceberg/src/transform/temporal.rs b/crates/iceberg/src/transform/temporal.rs index 1cd4d6a436..890e011ab1 100644 --- a/crates/iceberg/src/transform/temporal.rs +++ b/crates/iceberg/src/transform/temporal.rs @@ -661,7 +661,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (420034, 412007)"), + Some("name IN (412007, 420034)"), )?; fixture.assert_projection( @@ -807,7 +807,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (47, 46)"), + Some("name IN (46, 47)"), )?; fixture.assert_projection( @@ -879,7 +879,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (47, 46)"), + Some("name IN (46, 47)"), )?; fixture.assert_projection( @@ -951,7 +951,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (0, -1)"), + Some("name IN (-1, 0)"), )?; fixture.assert_projection( @@ -1023,7 +1023,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (575, 574)"), + Some("name IN (574, 575)"), )?; fixture.assert_projection( @@ -1094,7 +1094,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (-10, -9, -12, -11)"), + Some("name IN (-12, -11, -10, -9)"), )?; fixture.assert_projection( @@ -1240,7 +1240,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (1970-01-01, 1969-12-31)"), + Some("name IN (1969-12-31, 1970-01-01)"), )?; fixture.assert_projection( @@ -1314,7 +1314,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (2017-12-02, 2017-12-01)"), + Some("name IN (2017-12-01, 2017-12-02)"), )?; fixture.assert_projection( @@ -1388,7 +1388,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (1969-01-02, 1969-01-01, 1969-01-03)"), + Some("name IN (1969-01-01, 1969-01-02, 1969-01-03)"), )?; fixture.assert_projection( @@ -1462,7 +1462,7 @@ mod test { Datum::timestamp_from_str(value)?, Datum::timestamp_from_str(another)?, ]), - Some("name IN (2017-12-02, 2017-12-01)"), + Some("name IN (2017-12-01, 2017-12-02)"), )?; fixture.assert_projection( @@ -1740,7 +1740,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (-1, -12, -11, 0)"), + Some("name IN (-12, -11, -1, 0)"), )?; fixture.assert_projection( @@ -1808,7 +1808,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (575, 564)"), + Some("name IN (564, 575)"), )?; fixture.assert_projection( @@ -1876,7 +1876,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (-1, -12, -11, 0)"), + Some("name IN (-12, -11, -1, 0)"), )?; fixture.assert_projection( @@ -1944,7 +1944,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (575, 564)"), + Some("name IN (564, 575)"), )?; fixture.assert_projection( @@ -2012,7 +2012,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (0, -1)"), + Some("name IN (-1, 0)"), )?; fixture.assert_projection( @@ -2079,7 +2079,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (0, -1)"), + Some("name IN (-1, 0)"), )?; fixture.assert_projection( @@ -2147,7 +2147,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (47, 46)"), + Some("name IN (46, 47)"), )?; fixture.assert_projection( @@ -2215,7 +2215,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (0, -1)"), + Some("name IN (-1, 0)"), )?; fixture.assert_projection( @@ -2283,7 +2283,7 @@ mod test { Datum::date_from_str(value)?, Datum::date_from_str(another)?, ]), - Some("name IN (47, 46)"), + Some("name IN (46, 47)"), )?; fixture.assert_projection( diff --git a/crates/iceberg/src/transform/truncate.rs b/crates/iceberg/src/transform/truncate.rs index 4fac48f7d5..cf7e3485e6 100644 --- a/crates/iceberg/src/transform/truncate.rs +++ b/crates/iceberg/src/transform/truncate.rs @@ -469,7 +469,7 @@ mod test { Datum::decimal_from_str(curr)?, Datum::decimal_from_str(next)?, ]), - Some("name IN (10000, 10100, 9900)"), + Some("name IN (9900, 10000, 10100)"), )?; fixture.assert_projection( @@ -524,7 +524,7 @@ mod test { Datum::long(value), Datum::long(value + 1), ]), - Some("name IN (100, 90)"), + Some("name IN (90, 100)"), )?; fixture.assert_projection( @@ -579,7 +579,7 @@ mod test { Datum::long(value), Datum::long(value + 1), ]), - Some("name IN (100, 90)"), + Some("name IN (90, 100)"), )?; fixture.assert_projection( @@ -634,7 +634,7 @@ mod test { Datum::int(value), Datum::int(value + 1), ]), - Some("name IN (100, 90)"), + Some("name IN (90, 100)"), )?; fixture.assert_projection( @@ -689,7 +689,7 @@ mod test { Datum::int(value), Datum::int(value + 1), ]), - Some("name IN (100, 90)"), + Some("name IN (90, 100)"), )?; fixture.assert_projection( From 9f1cc288e66478b35c72238ac212f7f1661e207b Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Mon, 6 Jul 2026 16:57:35 +0800 Subject: [PATCH 4/5] chore: update public API snapshot Co-authored-by: Codex --- crates/iceberg/public-api.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 11ef635e0f..039772267d 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1564,6 +1564,7 @@ pub iceberg::spec::PrimitiveType::Timestamp pub iceberg::spec::PrimitiveType::TimestampNs pub iceberg::spec::PrimitiveType::Timestamptz pub iceberg::spec::PrimitiveType::TimestamptzNs +pub iceberg::spec::PrimitiveType::Unknown pub iceberg::spec::PrimitiveType::Uuid impl iceberg::spec::PrimitiveType pub fn iceberg::spec::PrimitiveType::compatible(&self, literal: &iceberg::spec::PrimitiveLiteral) -> bool From b062b0e6ef3a179ae0ac17f541598c2a22ac57eb Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Mon, 6 Jul 2026 17:23:08 +0800 Subject: [PATCH 5/5] fix: address PR review comments Co-authored-by: Codex --- crates/iceberg/src/expr/predicate.rs | 12 ++++++--- crates/iceberg/src/spec/schema/mod.rs | 39 ++++++++++++++++++++------- 2 files changed, 37 insertions(+), 14 deletions(-) diff --git a/crates/iceberg/src/expr/predicate.rs b/crates/iceberg/src/expr/predicate.rs index d99ed36674..f4257d2220 100644 --- a/crates/iceberg/src/expr/predicate.rs +++ b/crates/iceberg/src/expr/predicate.rs @@ -310,12 +310,16 @@ impl Bind for SetExpression { impl Display for SetExpression { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - let mut literals = self.literals.iter().collect_vec(); - literals.sort_by(|left, right| { + let mut literals = self + .literals + .iter() + .map(|literal| (literal, literal.to_string())) + .collect_vec(); + literals.sort_by(|(left, left_str), (right, right_str)| { left.partial_cmp(right) - .unwrap_or_else(|| left.to_string().cmp(&right.to_string())) + .unwrap_or_else(|| left_str.cmp(right_str)) }); - let mut literal_strs = literals.into_iter().map(ToString::to_string); + let mut literal_strs = literals.into_iter().map(|(_, literal_str)| literal_str); write!(f, "{} {} ({})", self.term, self.op, literal_strs.join(", ")) } diff --git a/crates/iceberg/src/spec/schema/mod.rs b/crates/iceberg/src/spec/schema/mod.rs index 2e49e6d15a..5a15a85633 100644 --- a/crates/iceberg/src/spec/schema/mod.rs +++ b/crates/iceberg/src/spec/schema/mod.rs @@ -195,13 +195,15 @@ impl SchemaBuilder { Type::Primitive(PrimitiveType::Unknown) => { ensure_data_valid!( !field.required, - "Field {} cannot be required because unknown type must be optional", - field.name + "Field {} (id {}) cannot be required because unknown type must be optional", + field.name, + field.id ); ensure_data_valid!( field.initial_default.is_none() && field.write_default.is_none(), - "Field {} cannot have non-null defaults because unknown type requires null defaults", - field.name + "Field {} (id {}) cannot have non-null defaults because unknown type requires null defaults", + field.name, + field.id ); } Type::Struct(struct_type) => { @@ -1370,11 +1372,9 @@ table { ]) .build() .unwrap_err(); - assert!( - required_error - .message() - .contains("unknown type must be optional") - ); + assert!(required_error.message().contains( + "Field empty (id 1) cannot be required because unknown type must be optional" + )); let default_error = Schema::builder() .with_schema_id(1) @@ -1388,7 +1388,26 @@ table { assert!( default_error .message() - .contains("unknown type requires null defaults") + .contains("Field empty (id 1) cannot have non-null defaults because unknown type requires null defaults") ); + + let nested_error = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::optional( + 1, + "items", + Type::List(ListType::new( + NestedField::list_element(2, Primitive(PrimitiveType::Unknown), true) + .into(), + )), + ) + .into(), + ]) + .build() + .unwrap_err(); + assert!(nested_error.message().contains( + "Field element (id 2) cannot be required because unknown type must be optional" + )); } }