nailo2c commented on code in PR #69234:
URL: https://github.com/apache/airflow/pull/69234#discussion_r3558153323
##########
providers/google/src/airflow/providers/google/cloud/openlineage/mixins.py:
##########
@@ -139,6 +176,61 @@ 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_id: str,
+ 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
+
+ from airflow.providers.openlineage.api.sql import emit_query_lineage
+ 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
+ start_time = self._get_bigquery_job_datetime(child_job_properties,
"startTime")
+ end_time = self._get_bigquery_job_datetime(child_job_properties,
"endTime")
+ emit_query_lineage(
Review Comment:
Nice idea, I'll try to implement 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]
