Skip to content

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

Closed

Description

@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.

Activity

  1. added
    kind:bugThis is a clearly a bug
    kind:metaHigh-level information important to the community
    priority:highHigh priority bug that should be patched quickly but does not require immediate new release
    on Apr 23, 2025
  2. self-assigned this
    on Apr 23, 2025
  3. added this to the Airflow 3.0.1 milestone on Apr 23, 2025
  4. bbovenzi commented on Apr 23, 2025

    @bbovenzi
    Contributor

    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?
    Image
  5. dstandish commented on Apr 29, 2025

    @dstandish
    ContributorAuthor

    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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

affected_version:3.0Issues Reported for 3.0area:backfillSpecifically for backfill relatedarea:corekind:bugThis is a clearly a bugkind:metaHigh-level information important to the communitypriority:highHigh priority bug that should be patched quickly but does not require immediate new release

Type

No type

Projects

No projects

    Milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions