feat(spec): add unknown primitive type support

crates/catalog/glue/src/schema.rs

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

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

@@ -1655,6 +1655,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

crates/iceberg/src/arrow/schema.rs

@@ -191,6 +191,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
@@ -509,6 +510,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))
@@ -697,6 +699,7 @@ impl SchemaVisitor for ToArrowSchemaConverter {
 
     fn primitive(&mut self, p: &PrimitiveType) -> Result<ArrowSchemaOrFieldOrType> {
         match p {
+            PrimitiveType::Unknown => Ok(ArrowSchemaOrFieldOrType::Type(DataType::Null)),
             PrimitiveType::Boolean => Ok(ArrowSchemaOrFieldOrType::Type(DataType::Boolean)),
             PrimitiveType::Int => Ok(ArrowSchemaOrFieldOrType::Type(DataType::Int32)),
             PrimitiveType::Long => Ok(ArrowSchemaOrFieldOrType::Type(DataType::Int64)),
@@ -1209,6 +1212,7 @@ pub(crate) fn primitive_type_to_arrow_type_with_ree(primitive_type: &PrimitiveTy
     };
 
     match primitive_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),
@@ -2227,6 +2231,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()
@@ -636,6 +637,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])))
         }
@@ -943,7 +945,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)),
 
         // --- Catch-all null arm: use arrow-rs new_null_array for any remaining DataType ---
         (dt, None) => new_null_array(dt, num_rows),
@@ -1866,6 +1868,26 @@ mod test {
         }
     }
 
+    #[test]
+    fn test_create_null_array_rejects_non_null_literal() {
+        let literal = Some(PrimitiveLiteral::Int(1));
+
+        assert!(create_primitive_array_single_element(&DataType::Null, &literal).is_err());
+        assert!(create_primitive_array_repeated(&DataType::Null, &literal, 2).is_err());
+        assert_eq!(
+            create_primitive_array_single_element(&DataType::Null, &None)
+                .unwrap()
+                .len(),
+            1
+        );
+        assert_eq!(
+            create_primitive_array_repeated(&DataType::Null, &None, 2)
+                .unwrap()
+                .len(),
+            2
+        );
+    }
+
     #[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 && !is_avro_null(&field_schema) {
             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 && !is_avro_null(&field_schema) {
             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 && !is_avro_null(&value_field_schema) {
             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,
@@ -311,6 +312,10 @@ pub(crate) fn avro_decimal_schema(precision: usize, scale: usize) -> Result<Avro
 }
 
 fn avro_optional(avro_schema: AvroSchema) -> Result<AvroSchema> {
+    if is_avro_null(&avro_schema) {
+        return Ok(AvroSchema::Null);
+    }
+
     Ok(AvroSchema::Union(UnionSchema::new(vec![
         AvroSchema::Null,
         avro_schema,
@@ -324,6 +329,14 @@ fn is_avro_optional(avro_schema: &AvroSchema) -> bool {
     }
 }
 
+fn is_avro_null(avro_schema: &AvroSchema) -> bool {
+    matches!(avro_schema, AvroSchema::Null)
+}
+
+fn unwrap_or_unknown(field_type: Option<Type>) -> Type {
+    field_type.unwrap_or(Type::Primitive(PrimitiveType::Unknown))
+}
+
 /// Post order avro schema visitor.
 pub(crate) trait AvroSchemaVisitor {
     type T;
@@ -447,10 +460,10 @@ 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) || is_avro_null(&avro_field.schema);
 
-            let mut field =
-                NestedField::new(field_id, &avro_field.name, field_type.unwrap(), !optional);
+            let field_type = unwrap_or_unknown(field_type);
+            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);
@@ -482,18 +495,19 @@ impl AvroSchemaVisitor for AvroSchemaToSchema {
         }
 
         if options.len() == 1 {
-            Ok(Some(options.remove(0).unwrap()))
+            Ok(Some(unwrap_or_unknown(options.remove(0))))
         } 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 = unwrap_or_unknown(item);
         let element_field = NestedField::list_element(
             element_field_id,
-            item.unwrap(),
-            !is_avro_optional(&array.items),
+            item,
+            !is_avro_optional(&array.items) && !is_avro_null(&array.items),
         )
         .into();
         Ok(Some(Type::List(ListType { element_field })))
@@ -504,10 +518,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 = unwrap_or_unknown(value);
         let value_field = NestedField::map_value_element(
             value_field_id,
-            value.unwrap(),
-            !is_avro_optional(&map.types),
+            value,
+            !is_avro_optional(&map.types) && !is_avro_null(&map.types),
         );
         Ok(Some(Type::Map(MapType {
             key_field: key_field.into(),
@@ -557,12 +572,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 = unwrap_or_unknown(value);
         let key_id = Self::get_element_id_from_attributes(
             &array.fields[0].custom_attributes,
             FIELD_ID_PROP,
@@ -575,7 +585,7 @@ 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) && !is_avro_null(&array.fields[1].schema),
         );
         Ok(Some(Type::Map(MapType {
             key_field: key_field.into(),
@@ -659,6 +669,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!(is_avro_null(&record.fields[0].schema));
+        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![