kacpermuda commented on code in PR #69234:
URL: https://github.com/apache/airflow/pull/69234#discussion_r3569948641
##########
providers/google/src/airflow/providers/google/cloud/openlineage/mixins.py:
##########
@@ -139,6 +171,76 @@ def get_openlineage_facets_on_complete(self, _):
job_facets={"sql":
SQLJobFacet(query=SQLParser.normalize_sql(self.sql))} if self.sql else {},
)
+ def _emit_child_query_lineage(
+ self,
+ *,
+ task_instance,
+ child_index: int,
+ child_job_properties: dict,
+ inputs: list[InputDataset | Dataset],
+ outputs: list[OutputDataset | Dataset],
+ ) -> None:
+ if task_instance is None:
+ self.log.debug("No task instance available. Skipping BigQuery
child job OpenLineage event.") # type: ignore[attr-defined]
+ return
+
+ try:
+ from airflow.providers.openlineage.api.sql import
emit_query_lineage
+ except ImportError:
+ self.log.debug( # type: ignore[attr-defined]
+ "The emit_query_lineage API requires
apache-airflow-providers-openlineage>=2.16.0. "
+ "Skipping BigQuery child job OpenLineage event."
+ )
+ return
+
+ from airflow.providers.openlineage.sqlparser import SQLParser
+
+ child_query = get_from_nullable_chain(child_job_properties,
["configuration", "query", "query"])
+ job_facets = {"sql":
SQLJobFacet(query=SQLParser.normalize_sql(child_query))} if child_query else
None
+ error_result = get_from_nullable_chain(child_job_properties,
["status", "errorResult"])
+ start_time = self._get_bigquery_job_datetime(child_job_properties,
"startTime")
+ end_time = self._get_bigquery_job_datetime(child_job_properties,
"endTime")
Review Comment:
nit: If one fails and returns None and the other succeeds, the
`emit_query_lineage` will fill in the failed one with now(), which may cause
the start > end or end > start, so we can probably guard passing timestamps to
the function only if both succeed. This is a nit, non blocking, just a small
improvement idea
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
