You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "mobuchowski (via GitHub)" <gi...@apache.org> on 2026/07/20 13:31:02 UTC
[PR] Link dbt Cloud runs to the triggering Airflow task via OpenLineage [airflow]
mobuchowski opened a new pull request, #70131:
URL: https://github.com/apache/airflow/pull/70131
dbt Cloud does not natively propagate the OpenLineage parent run of the Airflow task that triggered a job, so a triggered run has no stored link back to that task. Any tool that reads dbt Cloud runs on their own therefore sees them as isolated jobs, with no way to attach them to the Airflow run that started them. This adds an `openlineage_inject_parent_job_info` option to `DbtCloudRunJobOperator`. When enabled, it writes the task's OpenLineage parent and root run identifiers, as JSON, into the triggered run's `cause` field, so a consumer reading dbt Cloud runs can parse them and rebuild the `ParentRunFacet` that links the run to its Airflow task.
It is conceptually the same thing we've added in #44477 #64513 or #64807 but using different, dbt cloud specific mechanism.
The option defaults to the existing `openlineage.spark_inject_parent_job_info` config value, matching the convention already used by the Glue and EMR operators. Because dbt Cloud limits `cause` to 255 characters, the payload is trimmed to fit: the `parent` identifiers are always kept, the `root` block is dropped if the two together are too long, and injection is skipped if even the parent alone does not fit. When injection is enabled the `cause` is replaced entirely with these identifiers rather than the human-readable trigger reason, since the correlation is the point of the option.
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Opus 4.8) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Link dbt Cloud runs to the triggering Airflow task via OpenLineage [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski merged PR #70131:
URL: https://github.com/apache/airflow/pull/70131
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Link dbt Cloud runs to the triggering Airflow task via OpenLineage [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #70131:
URL: https://github.com/apache/airflow/pull/70131#discussion_r3622969609
##########
providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py:
##########
@@ -97,6 +99,62 @@ def _get_parent_run_metadata(task_instance):
)
+def _get_openlineage_parent_facet_dict(task_instance) -> dict:
+ """Build the OpenLineage ``ParentRunFacet`` payload (parent and root) for a task instance."""
+ metadata = _get_parent_run_metadata(task_instance)
+ return {
+ "parent": {
+ "run": {"runId": metadata.run_id},
+ "job": {"namespace": metadata.job_namespace, "name": metadata.job_name},
+ },
+ "root": {
+ "run": {"runId": metadata.root_parent_run_id},
+ "job": {
+ "namespace": metadata.root_parent_job_namespace,
+ "name": metadata.root_parent_job_name,
+ },
+ },
+ }
+
+
+def inject_parent_job_information_into_dbt_cloud_cause(cause: str | None, task_instance) -> str | None:
+ """
+ Replace the dbt Cloud run ``cause`` with OpenLineage parent job information as JSON.
+
+ This serializes the Airflow task's OpenLineage parent (and root) run identifiers into the
+ triggered run's ``cause`` field. A consumer that reads dbt Cloud runs can parse this JSON and
+ attach a ``ParentRunFacet``, linking the dbt Cloud run back to the Airflow task that triggered it.
+
+ The dbt Cloud ``cause`` is limited to ``DBT_CAUSE_MAX_LENGTH`` characters. The essential ``parent``
+ link is always kept; the ``root`` block is dropped if the two together do not fit. If even the
+ parent-only payload does not fit, the original ``cause`` is returned unchanged.
+ """
+ try:
+ facet = _get_openlineage_parent_facet_dict(task_instance)
+ except ImportError:
+ log.warning(
+ "Could not import OpenLineage provider. Skipping the injection of OpenLineage "
+ "parent job information into the dbt Cloud run cause."
+ )
+ return cause
+
+ candidates = [
+ {"parent": facet["parent"], "root": facet["root"]},
+ {"parent": facet["parent"]},
+ ]
Review Comment:
I know that every character counts since the limit is low, but maybe we should add encapsulating `{"openlineage": {"parent"...}}` dict? Without the word `openlineage`, parent and root can be confusing for reader. But, if somebody sets it to True, they usually know what they're doing. Something to consider.
##########
providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py:
##########
@@ -219,6 +233,12 @@ def execute(self, context: Context):
f"Triggered via Apache Airflow by task {self.task_id!r} in the {self.dag.dag_id} DAG."
)
+ if self.openlineage_inject_parent_job_info:
+ self.log.info("Injecting OpenLineage parent job information into dbt Cloud run cause.")
Review Comment:
```suggestion
self.log.info("Replacing dbt Cloud run cause with OpenLineage parent job information.")
```
Let's be explicit here, since we replace and not inject (because the char limit on cause is so low)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Link dbt Cloud runs to the triggering Airflow task via OpenLineage [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #70131:
URL: https://github.com/apache/airflow/pull/70131#discussion_r3628764479
##########
providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py:
##########
@@ -97,6 +99,62 @@ def _get_parent_run_metadata(task_instance):
)
+def _get_openlineage_parent_facet_dict(task_instance) -> dict:
+ """Build the OpenLineage ``ParentRunFacet`` payload (parent and root) for a task instance."""
+ metadata = _get_parent_run_metadata(task_instance)
+ return {
+ "parent": {
+ "run": {"runId": metadata.run_id},
+ "job": {"namespace": metadata.job_namespace, "name": metadata.job_name},
+ },
+ "root": {
+ "run": {"runId": metadata.root_parent_run_id},
+ "job": {
+ "namespace": metadata.root_parent_job_namespace,
+ "name": metadata.root_parent_job_name,
+ },
+ },
+ }
+
+
+def inject_parent_job_information_into_dbt_cloud_cause(cause: str | None, task_instance) -> str | None:
+ """
+ Replace the dbt Cloud run ``cause`` with OpenLineage parent job information as JSON.
+
+ This serializes the Airflow task's OpenLineage parent (and root) run identifiers into the
+ triggered run's ``cause`` field. A consumer that reads dbt Cloud runs can parse this JSON and
+ attach a ``ParentRunFacet``, linking the dbt Cloud run back to the Airflow task that triggered it.
+
+ The dbt Cloud ``cause`` is limited to ``DBT_CAUSE_MAX_LENGTH`` characters. The essential ``parent``
+ link is always kept; the ``root`` block is dropped if the two together do not fit. If even the
+ parent-only payload does not fit, the original ``cause`` is returned unchanged.
+ """
+ try:
+ facet = _get_openlineage_parent_facet_dict(task_instance)
+ except ImportError:
+ log.warning(
+ "Could not import OpenLineage provider. Skipping the injection of OpenLineage "
+ "parent job information into the dbt Cloud run cause."
+ )
+ return cause
+
+ candidates = [
+ {"parent": facet["parent"], "root": facet["root"]},
+ {"parent": facet["parent"]},
+ ]
Review Comment:
I think we're too constrained here unfortunately :(
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org