Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

MINOR: Tableau Capture SQL in Lineage #18925

Merged
merged 1 commit into from
Dec 5, 2024
Merged

MINOR: Tableau Capture SQL in Lineage #18925

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ingestion/src/metadata/ingestion/source/dashboard/dashboard_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -479,6 +479,7 @@ def _get_add_lineage_request(
to_entity: Union[Dashboard, DashboardDataModel],
from_entity: Union[Table, DashboardDataModel, Dashboard],
column_lineage: List[ColumnLineage] = None,
sql: Optional[str] = None,
) -> Optional[Either[AddLineageRequest]]:
if from_entity and to_entity:
return Either(
Expand All @@ -494,6 +495,7 @@ def _get_add_lineage_request(
),
lineageDetails=LineageDetails(
source=LineageSource.DashboardLineage,
sqlQuery=sql,
columnsLineage=column_lineage,
),
)
Expand Down
28 changes: 19 additions & 9 deletions ingestion/src/metadata/ingestion/source/dashboard/tableau/metadata.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@
ChartUrl,
DataSource,
DatasourceField,
TableAndQuery,
TableauDashboard,
TableauTag,
UpstreamTable,
Expand Down Expand Up @@ -149,7 +150,7 @@ def get_owner_ref(
return None

@staticmethod
def _get_data_models_tags(dataModels: [DataSource]) -> Set[str]:
def _get_data_models_tags(dataModels: List[DataSource]) -> Set[str]:
"""
Get the tags from the data model in the upstreamDatasources
"""
Expand Down Expand Up @@ -454,14 +455,18 @@ def _get_table_datamodel_lineage(
}
for table in datamodel.upstreamTables or []:
om_tables = self._get_database_tables(db_service_entity, table)
for om_table in om_tables or []:
for om_table_and_query in om_tables or []:
column_lineage = self._get_column_lineage(
table, om_table, upstream_data_model_entity, upstream_col_set
table,
om_table_and_query.table,
upstream_data_model_entity,
upstream_col_set,
)
yield self._get_add_lineage_request(
to_entity=upstream_data_model_entity,
from_entity=om_table,
from_entity=om_table_and_query.table,
column_lineage=column_lineage,
sql=om_table_and_query.query,
)
except Exception as err:
yield Either(
Expand Down Expand Up @@ -698,7 +703,7 @@ def close(self):

def _get_table_entities_from_api(
self, db_service_entity: DatabaseService, table: UpstreamTable
) -> Optional[List[Table]]:
) -> Optional[List[TableAndQuery]]:
"""
In case we get the table details from the Graphql APIs we process them
"""
Expand Down Expand Up @@ -734,15 +739,15 @@ def _get_table_entities_from_api(
fqn=table_fqn,
)
if table_entity:
return [table_entity]
return [TableAndQuery(table=table_entity)]
except Exception as exc:
logger.debug(traceback.format_exc())
logger.warning(f"Error to get tables for lineage using GraphQL Apis: {exc}")
return None

def _get_table_entities_from_query(
self, db_service_entity: DatabaseService, table: UpstreamTable
) -> Optional[List[Table]]:
) -> Optional[List[TableAndQuery]]:
"""
In case we get the table details from the Graphql APIs we process them
"""
Expand Down Expand Up @@ -778,7 +783,12 @@ def _get_table_entities_from_query(
database_schema=schema_name,
table=table_name,
)
tables_list.extend(from_entities)
tables_list.extend(
[
TableAndQuery(table=table, query=custom_sql_table.query)
for table in from_entities
]
)

except Exception as exc:
logger.debug(traceback.format_exc())
Expand All @@ -787,7 +797,7 @@ def _get_table_entities_from_query(

def _get_database_tables(
self, db_service_entity: DatabaseService, table: UpstreamTable
) -> Optional[List[Table]]:
) -> Optional[List[TableAndQuery]]:
"""
Get the table entities for lineage
"""
Expand Down
10 changes: 10 additions & 0 deletions ingestion/src/metadata/ingestion/source/dashboard/tableau/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from pydantic import BaseModel, ConfigDict, Field, validator

from metadata.generated.schema.entity.data.chart import ChartType
from metadata.generated.schema.entity.data.table import Table


class TableauBaseModel(BaseModel):
Expand Down Expand Up @@ -172,3 +173,12 @@ class TableauDashboard(TableauBaseModel):
webpageUrl: Optional[str] = None
charts: Optional[List[TableauChart]] = None
dataModels: List[DataSource] = []


class TableAndQuery(BaseModel):
"""
Wrapper class for Table entity and associated Query for lineage
"""

table: Table
query: Optional[str] = None
Loading