Merge pull request #2 from datahub-project/feature/iceberg_source

cccs-eric · web-flow · commit 299a7f86344e · 2022-05-12T10:01:12.000-04:00
Iceberg: Use logical type in the contained types for ArrayTypeClass &amp; MapTypeClass + fix unit-tests.
diff --git a/metadata-ingestion/src/datahub/ingestion/extractor/schema_util.py b/metadata-ingestion/src/datahub/ingestion/extractor/schema_util.py
@@ -129,8 +129,17 @@ def __init__(self, is_key_schema: bool, default_nullable: bool = False) -> None:
         }
 
     @staticmethod
-    def _get_type_name(avro_schema: avro.schema.Schema) -> str:
-        return str(getattr(avro_schema.type, "type", avro_schema.type))
+    def _get_type_name(
+        avro_schema: avro.schema.Schema, logical_if_present: bool = False
+    ) -> str:
+        logical_type_name: Optional[str] = None
+        if logical_if_present:
+            logical_type_name = getattr(
+                avro_schema, "logical_type", None
+            ) or avro_schema.props.get("logicalType")
+        return logical_type_name or str(
+            getattr(avro_schema.type, "type", avro_schema.type)
+        )
 
     @staticmethod
     def _get_column_type(
@@ -151,15 +160,17 @@ def _get_column_type(
             avro_schema, avro.schema.ArraySchema
         ):
             dt.type.nestedType = [
-                AvroToMceSchemaConverter._get_type_name(avro_schema.items)
+                AvroToMceSchemaConverter._get_type_name(
+                    avro_schema.items, logical_if_present=True
+                )
             ]
         elif isinstance(dt.type, MapTypeClass) and isinstance(
             avro_schema, avro.schema.MapSchema
         ):
             # Avro map's key is always a string. See: https://avro.apache.org/docs/current/spec.html#Maps
             dt.type.keyType = "string"
             dt.type.valueType = AvroToMceSchemaConverter._get_type_name(
-                avro_schema.values
+                avro_schema.values, logical_if_present=True
             )
         return dt
 
diff --git a/metadata-ingestion/tests/unit/test_iceberg.py b/metadata-ingestion/tests/unit/test_iceberg.py
@@ -192,10 +192,10 @@ def test_iceberg_primitive_type_to_schema_field(
             IcebergTypes.TimestampType.without_timezone(),
             "timestamp-micros",
         ),
-        (IcebergTypes.TimeType.get(), "timestamp-micros"),
+        (IcebergTypes.TimeType.get(), "time-micros"),
         (
             IcebergTypes.UUIDType.get(),
-            "string",
+            "uuid",
         ),
     ],
 )
@@ -250,10 +250,10 @@ def test_iceberg_list_to_schema_field(
             IcebergTypes.TimestampType.without_timezone(),
             "timestamp-micros",
         ),
-        (IcebergTypes.TimeType.get(), "timestamp-micros"),
+        (IcebergTypes.TimeType.get(), "time-micros"),
         (
             IcebergTypes.UUIDType.get(),
-            "string",
+            "uuid",
         ),
     ],
 )
