From 44645b9650ea9b79d4493b9ddca3502654657bd0 Mon Sep 17 00:00:00 2001
From: Bumyu <desleydelits@Gmail.com>
Date: Wed, 2 Oct 2024 05:59:37 +0200
Subject: [PATCH] fix(ingest/elasticsearch): detect sub-properties in 'nested'
 type mapping (#11338)

Co-authored-by: Lawrence De Spiegeleire <lawrence.despiegeleire@pauwelconsulting.com>
---
 .../ingestion/source/elastic_search.py        | 20 +++++++++----------
 1 file changed, 10 insertions(+), 10 deletions(-)

diff --git a/metadata-ingestion/src/datahub/ingestion/source/elastic_search.py b/metadata-ingestion/src/datahub/ingestion/source/elastic_search.py
index 8bda5db9a379a0..aa5913f5dc66b1 100644
--- a/metadata-ingestion/src/datahub/ingestion/source/elastic_search.py
+++ b/metadata-ingestion/src/datahub/ingestion/source/elastic_search.py
@@ -138,31 +138,31 @@ def _get_schema_fields(
         for columnName, column in elastic_schema_dict.items():
             elastic_type: Optional[str] = column.get("type")
             nested_props: Optional[Dict[str, Any]] = column.get(PROPERTIES)
-            if elastic_type is not None:
-                self._prefix_name_stack.append(f"[type={elastic_type}].{columnName}")
-                schema_field_data_type = self.get_column_type(elastic_type)
+            if nested_props:
+                self._prefix_name_stack.append(f"[type={PROPERTIES}].{columnName}")
                 schema_field = SchemaField(
                     fieldPath=self._get_cur_field_path(),
-                    nativeDataType=elastic_type,
-                    type=schema_field_data_type,
+                    nativeDataType=PROPERTIES,
+                    type=SchemaFieldDataTypeClass(RecordTypeClass()),
                     description=None,
                     nullable=True,
                     recursive=False,
                 )
                 yield schema_field
+                yield from self._get_schema_fields(nested_props)
                 self._prefix_name_stack.pop()
-            elif nested_props:
-                self._prefix_name_stack.append(f"[type={PROPERTIES}].{columnName}")
+            elif elastic_type is not None:
+                self._prefix_name_stack.append(f"[type={elastic_type}].{columnName}")
+                schema_field_data_type = self.get_column_type(elastic_type)
                 schema_field = SchemaField(
                     fieldPath=self._get_cur_field_path(),
-                    nativeDataType=PROPERTIES,
-                    type=SchemaFieldDataTypeClass(RecordTypeClass()),
+                    nativeDataType=elastic_type,
+                    type=schema_field_data_type,
                     description=None,
                     nullable=True,
                     recursive=False,
                 )
                 yield schema_field
-                yield from self._get_schema_fields(nested_props)
                 self._prefix_name_stack.pop()
             else:
                 # Unexpected! Log a warning.