yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-12-13 11:06:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ingest/elasticsearch): detect sub-properties in 'nested' type mapping (#11338)

Browse Source 
Co-authored-by: Lawrence De Spiegeleire <lawrence.despiegeleire@pauwelconsulting.com>
This commit is contained in:
Bumyu 2024-10-02 05:59:37 +02:00 committed by GitHub
parent a87c123611
commit 44645b9650
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
metadata-ingestion/src/datahub/ingestion/source/elastic_search.py
View File

@ -138,20 +138,7 @@ class ElasticToSchemaFieldConverter:
for columnName, column in elastic_schema_dict.items(): for columnName, column in elastic_schema_dict.items():
elastic_type: Optional[str] = column.get("type") elastic_type: Optional[str] = column.get("type")
nested_props: Optional[Dict[str, Any]] = column.get(PROPERTIES) nested_props: Optional[Dict[str, Any]] = column.get(PROPERTIES)
if elastic_type is not None: if nested_props:
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=elastic_type,
type=schema_field_data_type,
description=None,
nullable=True,
recursive=False,
)
yield schema_field
self._prefix_name_stack.pop()
elif nested_props:
self._prefix_name_stack.append(f"[type={PROPERTIES}].{columnName}") self._prefix_name_stack.append(f"[type={PROPERTIES}].{columnName}")
schema_field = SchemaField( schema_field = SchemaField(
fieldPath=self._get_cur_field_path(), fieldPath=self._get_cur_field_path(),
@ -164,6 +151,19 @@ class ElasticToSchemaFieldConverter:
yield schema_field yield schema_field
yield from self._get_schema_fields(nested_props) yield from self._get_schema_fields(nested_props)
self._prefix_name_stack.pop() self._prefix_name_stack.pop()
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=elastic_type,
type=schema_field_data_type,
description=None,
nullable=True,
recursive=False,
)
yield schema_field
self._prefix_name_stack.pop()
else: else:
# Unexpected! Log a warning. # Unexpected! Log a warning.
logger.warning( logger.warning(