yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-19 06:28:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

MINOR: Lineage handle copy queries being skipped (#14855)

Browse Source
This commit is contained in:
Mayur Singal 2024-01-25 10:15:32 +05:30 committed by GitHub
parent 70316b9680
commit 17fb2cabca
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 52 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
ingestion/src/metadata/ingestion/lineage/parser.py
View File

@ -98,24 +98,30 @@ class LineageParser:
"""
Get a list of intermediate tables
"""
if self.parser:
# These are @lazy_property, not properly being picked up by IDEs. Ignore the warning
return self.retrieve_tables(self.parser.intermediate_tables)
return []
@cached_property
def source_tables(self) -> List[Table]:
"""
Get a list of source tables
"""
if self.parser:
# These are @lazy_property, not properly being picked up by IDEs. Ignore the warning
return self.retrieve_tables(self.parser.source_tables)
return []
@cached_property
def target_tables(self) -> List[Table]:
"""
Get a list of target tables
"""
if self.parser:
# These are @lazy_property, not properly being picked up by IDEs. Ignore the warning
return self.retrieve_tables(self.parser.target_tables)
return []
# pylint: disable=protected-access
@cached_property
@ -124,6 +130,8 @@ class LineageParser:
Get a list of tuples of column lineage
"""
column_lineage = []
if self.parser is None:
return []
try:
if self.parser._dialect == SQLPARSE_DIALECT:
return self.parser.get_column_lineage()
@ -331,6 +339,8 @@ class LineageParser:
:return: for each table name, list all joins against other tables
"""
join_data = defaultdict(list)
if self.parser is None:
return join_data
# These are @lazy_property, not properly being picked up by IDEs. Ignore the warning
for statement in self.parser.statements():
self.stateful_add_joins_from_statement(join_data, sql_statement=statement)
@ -378,7 +388,11 @@ class LineageParser:
@staticmethod
def _evaluate_best_parser(
query: str, dialect: Dialect, timeout_seconds: int
) -> LineageRunner:
) -> Optional[LineageRunner]:
if query is None:
return None
@timeout(seconds=timeout_seconds)
def get_sqlfluff_lineage_runner(qry: str, dlct: str) -> LineageRunner:
lr_dialect = LineageRunner(qry, dialect=dlct)

29
ingestion/tests/unit/test_query_parser.py
View File

@ -226,7 +226,6 @@ class QueryParserTests(TestCase):
;
"""
expected_tables = {"testdb.public.users"}
expected_lineage = [
(
Column("testdb.public.users.id"),
@ -329,3 +328,31 @@ class QueryParserTests(TestCase):
parser.column_lineage,
expected_lineage,
)
def test_copy_query(self):
"""
Validate Copy query is skipped appropriately without any errors
"""
query = """COPY MY_TABLE col1,col2,col3
FROM 's3://bucket/schema/table.csv'
WITH CREDENTIALS ''
REGION 'US-east-2'
"""
expected_lineage = []
expected_tables = set()
parser = LineageParser(query)
tables = {str(table) for table in parser.involved_tables}
self.assertEqual(tables, expected_tables)
self.assertEqual(
parser.column_lineage,
expected_lineage,
)
parser = LineageParser(query, Dialect.MYSQL)
tables = {str(table) for table in parser.involved_tables}
self.assertEqual(tables, expected_tables)
self.assertEqual(
parser.column_lineage,
expected_lineage,
)