nailo2c commented on code in PR #69234:
URL: https://github.com/apache/airflow/pull/69234#discussion_r3585646317
##########
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:
Nice catch! Very useful, I'll fix it :)
--
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]
