openlineage: Add AirflowDagRunFacet to dag runEvents

[dolfinus]

closes: #40798

• Add AirflowDagRunFacet which copies AirflowRunFacet, but without fields task, taskInstance, taskUuid.
• Add the same facet to DAG run events with eventType=START.

This allows to collect data like dag tags, dag schedule interval, dagRun id and type on OpenLineage consumer side. Before that, this information was available only for task events.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[mobuchowski]

@dolfinus I think this should be a different facet or added to AirflowJobFacet that we attach to DagRun events.

[dolfinus]

Added to AirflowJobFacet that we attach to DagRun events.

DagRun information is a part of RunFacet, as it has fields like start_date, run_id, which are not JobFacet.

I think this should be a different facet

Honestly, I'd rather split these fields to different facets, like:

job:
  facets:
    airflow_dag: AirflowDagJobFacet(...)  # previously airflow:dag
    airflow_task: AirflowTaskJobFacet(...)  # previously airflow:task
run:
  facets:
    airflow_dagRun: AirflowDagRunFacet(...)  # previously airflow:dagRun
    airflow_taskInstance: AirflowTaskInstanceFacet(...)  # previously airflow:taskInstance
    # throw away taskUuuid, it just ruplicates runId

But this is not backwards compatible.

[mobuchowski]

Why not attach it to some new facet as AirflowDagRunFacet? Basically, AirflowRunFacet is tightly coupled to what we have during task instance execution, and don't think making it just a bag of possible things makes sense.

I'd rather have the same DagRun information attached to different facets in TaskInstance level events and DagRun events.

[dolfinus]

Why not attach it to some new facet as AirflowDagRunFacet?

What should be the key of this facet? "airflow" or something else? For OL consumer, having runEvents with facets stored with the same key, but having different fields for Task and DAG runs, is not very comfortable.

Should AirflowRunFacet be renamed to AirflowTaskRunFacet, for consistency?

[kacpermuda]

Why not attach it to some new facet as AirflowDagRunFacet?

What should be the key of this facet? "airflow" or something else? For OL consumer, having runEvents with facets stored with the same key, but having different fields for Task and DAG runs, is not very comfortable.

Should AirflowRunFacet be renamed to AirflowTaskRunFacet, for consistency?

We can think of a new name, that would not be confusing. I don't think we even could go with another run facet with airflow as key, as from spec perspective it would be a duplicate, there is no distinction between DAG and Task run facet from the spec perspective.

I think the renaming you propose would not be a big issue, as it would not change the spec, only the import, so we would have to keep old name for compatibility. Personally, I would avoid doing any changes to AirflowRunFacet if possible. I think renaming will not be necessary if we come up with a better name for the new one 😄

[dolfinus]

My current plan is:

• Mark AirflowRunFacet as deprecated, but keep it for backward compatibility (al least for several next releases).
• Add new facet airflow_task: AirflowTaskRunFacet which would just copy AirflowRunFacet, but with a new name and key.
• Add new facet airflow_dag: AirflowDagRunFacet which would implement changes for OpenLineage: Add AirflowRunFacet to DAG run events #40798.

What do you think?

[kacpermuda]

My current plan is:

• Mark AirflowRunFacet as deprecated, but keep it for backward compatibility (al least for several next releases).
• Add new facet airflow_task: AirflowTaskRunFacet which would just copy AirflowRunFacet, but with a new name and key.
• Add new facet airflow_dag: AirflowDagRunFacet which would implement changes for OpenLineage: Add AirflowRunFacet to DAG run events #40798.

What do you think?

I think i would proceed with the last point for now in this PR, and create a discussion for the first two so that we can decide if / how we want to make it less confusing for users. The AirflowRunFacet is the primary facet that users rely on, so we need to handle it with care. Before proceeding with any actions, we should have an in-depth discussion, as deprecating it without such a discussion would not be advisable. 😄

We can also discuss here what would be the best key for this new facet: airflowDag, airflowRun, airflowDagRun?

[dolfinus]

Updated implementation - reverted all the changes to AirflowRunFacet, added new facet airflow_dagRun: AirflowDagRunFacet. Using key airflow_dagRun instead of airflowDagRun as recommended by specification.

Regarding changing AirflowRunFacet name, key and probably content I've created discussion #40886.

[kacpermuda]

Overall LGTM, left some comments

[kacpermuda]

LGTM, left some comments