feat(spec): add unknown datatype support

crates/catalog/glue/src/schema.rs

@@ -157,6 +157,12 @@ impl SchemaVisitor for GlueSchemaBuilder {
 
     fn primitive(&mut self, p: &iceberg::spec::PrimitiveType) -> iceberg::Result<Self::T> {
         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(),

crates/catalog/hms/src/schema.rs

@@ -114,6 +114,12 @@ impl SchemaVisitor for HiveSchemaBuilder {
 
     fn primitive(&mut self, p: &iceberg::spec::PrimitiveType) -> iceberg::Result<String> {
         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(),

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
@@ -2313,6 +2314,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<Item = i32> + '_
 pub fn iceberg::spec::Schema::into_builder(self) -> iceberg::spec::SchemaBuilder
+pub fn iceberg::spec::Schema::min_format_version(&self) -> core::option::Option<iceberg::spec::FormatVersion>
 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

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<i32>) {
         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,

crates/iceberg/src/arrow/schema.rs

@@ -120,6 +120,7 @@ fn visit_type<V: ArrowSchemaVisitor>(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<Self::T> {
         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<ArrowSchemaOrFieldOrType> {
         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

crates/iceberg/src/arrow/value.rs

@@ -206,6 +206,7 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
 
     fn primitive(&mut self, p: &PrimitiveType, partner: &ArrayRef) -> Result<Vec<Option<Literal>>> {
         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<PrimitiveLiteral>,
 ) -> Result<ArrayRef> {
     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.

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<AvroSchemaOrField> {
         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<AvroSchemaOrField> {
         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<Avro
 }
 
 fn avro_optional(avro_schema: AvroSchema) -> Result<AvroSchema> {
+    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,18 +481,21 @@ 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()))
         }
     }
 
     fn array(&mut self, array: &ArraySchema, item: Option<Type>) -> Result<Self::T> {
         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![

crates/iceberg/src/expr/predicate.rs

@@ -310,7 +310,16 @@ impl<T: Bind> Bind for SetExpression<T> {
 
 impl<T: Display + Debug> Display for SetExpression<T> {
     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()
+            .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_str.cmp(right_str))
+        });
+        let mut literal_strs = literals.into_iter().map(|(_, literal_str)| literal_str);
 
         write!(f, "{} {} ({})", self.term, self.op, literal_strs.join(", "))
     }
@@ -1363,7 +1372,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 +1407,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 +1580,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]