Skip to content

Add unknown datatype support #2410

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
manuzhang wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Add unknown datatype support #2410

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions crates/catalog/glue/src/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
6 changes: 6 additions & 0 deletions crates/catalog/hms/src/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
2 changes: 2 additions & 0 deletions crates/iceberg/src/arrow/reader/projection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,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);
}
Expand Down Expand Up @@ -94,6 +95,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,
Expand Down
13 changes: 13 additions & 0 deletions crates/iceberg/src/arrow/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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))
}
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions crates/iceberg/src/arrow/value.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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, _) => Ok(Arc::new(arrow_array::NullArray::new(1))),
(DataType::Boolean, Some(PrimitiveLiteral::Boolean(v))) => {
Ok(Arc::new(BooleanArray::from(vec![*v])))
}
Expand Down
61 changes: 43 additions & 18 deletions crates/iceberg/src/avro/schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
}

Expand Down Expand Up @@ -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)?;
}

Expand All @@ -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)?;
}

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 })))
Expand All @@ -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(),
Expand Down Expand Up @@ -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,
Expand All @@ -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(),
Expand Down Expand Up @@ -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![
Expand Down
19 changes: 9 additions & 10 deletions crates/iceberg/src/expr/predicate.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,7 +310,12 @@ 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().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(", "))
}
Expand Down Expand Up @@ -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);
}

Expand Down Expand Up @@ -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);
}

Expand Down Expand Up @@ -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]
Expand Down
Loading
Loading