Handle Avro logical types fully specified inside the type.

metadata-ingestion/src/datahub/ingestion/extractor/schema_util.py

@@ -287,7 +287,15 @@ def emit(self) -> Generator[SchemaField, None, None]:
                     # Populate it with the simple native type for now.
                     nativeDataType=native_data_type,
                     type=self._converter._get_column_type(
-                        actual_schema.type, self._actual_schema.props.get("logicalType")
+                        actual_schema.type,
+                        (
+                            getattr(
+                                actual_schema, "logical_type", None
+                            )  # logicalType nested inside type
+                            or self._actual_schema.props.get(
+                                "logicalType"
+                            )  # bare logicalType
+                        ),
                     ),
                     description=description,
                     recursive=False,

metadata-ingestion/tests/unit/test_schema_util.py

@@ -1,8 +1,9 @@
+import json
 import logging
 import os
 import re
 from pathlib import Path
-from typing import List, Type
+from typing import Dict, List, Type
 
 import pytest
 
@@ -647,7 +648,7 @@ def test_key_schema_handling():
         assert f.isPartOfKey
 
 
-def test_logical_types():
+def test_logical_types_bare():
     schema: str = """
 {
     "type": "record",
@@ -689,6 +690,32 @@ def test_logical_types():
     assert expected_types == [type(field.type.type) for field in fields]
 
 
+def test_logical_types_fully_specified_in_type():
+    schema: Dict = {
+        "type": "record",
+        "name": "test",
+        "fields": [
+            {
+                "name": "name",
+                "type": {
+                    "type": "bytes",
+                    "logicalType": "decimal",
+                    "precision": 3,
+                    "scale": 2,
+                    "native_data_type": "decimal(3, 2)",
+                    "_nullable": True,
+                },
+            }
+        ],
+    }
+    fields: List[SchemaField] = avro_schema_to_mce_fields(
+        json.dumps(schema), default_nullable=True
+    )
+    assert len(fields) == 1
+    assert "[version=2.0].[type=test].[type=bytes].name" == fields[0].fieldPath
+    assert isinstance(fields[0].type.type, NumberTypeClass)
+
+
 def test_ignore_exceptions():
     malformed_schema: str = """
   "name": "event_ts",