Ensure that we clear bundle version when clearing a run -- for backfill or otherwise

[dstandish]

Body

When clearing an existing dag run, I think it makes sense to use the latest bundle version.

Let's make sure we do that, if we don't already.

cc @cmarteepants @bbovenzi

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[bbovenzi]

Testing locally.

Clearing a single task instance:

• task instance version number updates
• that version is added to the list of versions for the dag run

Clearing an entire dag run:

• task instances versions update
• dag run version is just added to the versions array on a dag run. I feel like we would want to only show the latest version on the dagun in this case?

[dstandish]

Testing locally.

Clearing a single task instance:

• task instance version number updates
• that version is added to the list of versions for the dag run

Clearing an entire dag run:

• task instances versions update
• dag run version is just added to the versions array on a dag run. I feel like we would want to only show the latest version on the dagun in this case?

Note that there is a difference between bundle version and DagVersion. DagVersion tracks changes to serialized dag structure, and is not 1-1 with bundle version.

Right now, the logic to return dag versions depends on whether there is a bundle version or not:

    @property
    def dag_versions(self) -> list[DagVersion]:
        """Return the DAG versions associated with the TIs of this DagRun."""
        # when the dag is in a versioned bundle, we keep the dag version fixed
        if self.bundle_version:
            return [self.created_dag_version]
        dag_versions = [
            dv
            for dv in dict.fromkeys(list(self._tih_dag_versions) + list(self._ti_dag_versions))
            if dv is not None
        ]
        sorted_ = sorted(dag_versions, key=lambda dv: dv.id)
        return sorted_

So, if you are using a versioned bundle, Airlfow will only ever consider the dag structure that was there when the dag run was created and if it changes in flight (e.g. if the structure is driven by a variable or something) you are out of luck.

And you will only ever have more than one dag version for a given dag run when the bundle is non-versioned (or if you disable bundle versioning for the dag, which is a feature also). And for such dags, there can be multiple versions. And, as you can see in that code snippet, we query all the TIs (including the latest "try" and those before) that were ever part of the dag run, to get the union of all dag versions -- and my understanding this is to serve the UI.