feat(dag_version): add soft-delete support and set dag_version FK to SET NULL

Author: sunank200

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-4cda55eb221ee0340749c8dd41af7603220d7b300e2e808d733239a86e0c2837
+c81c0f0c5e071b67a71b959a1a78f5b9357893c0b73efa28eaf890b8a2e299db

airflow-core/docs/img/airflow_erd.svg

@@ -39,7 +39,12 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``0242ac120002`` (head) | ``dfee8bd5d574`` | ``3.1.0``         | Change the Deadline column in the Deadline table from        |
+| ``1b2cd3e4f5a6`` (head) | ``03e36c7f30aa`` | ``3.1.0``         | Modify task_instance.dag_version_id FK ondelete behavior to  |
+|                         |                  |                   | SET NULL.                                                    |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``03e36c7f30aa``        | ``0242ac120002`` | ``3.1.0``         | Add soft-delete flag to DagVersion (is_active).              |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``0242ac120002``        | ``dfee8bd5d574`` | ``3.1.0``         | Change the Deadline column in the Deadline table from        |
 |                         |                  |                   | DateTime to UTC DateTime.                                    |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``dfee8bd5d574``        | ``29ce7909c52b`` | ``3.1.0``         | Add Deadline to Dag.                                         |

airflow-core/docs/migrations-ref.rst

@@ -191,6 +191,8 @@ def grid_data(
                 task_node_map_exclude and ti.task_id not in task_node_map_exclude
             ) or ti.state == TaskInstanceState.REMOVED:
                 continue
+            if ti.task_id not in task_node_map:
+                continue
 
             # Populate the Grouped Task Instances (All Task Instances except the Parent Task Instances)
             if ti.task_id in get_child_task_map(

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py

@@ -0,0 +1,57 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add soft-delete flag to DagVersion (is_active).
+
+Revision ID: 03e36c7f30aa
+Revises: 0242ac120002
+Create Date: 2025-05-20 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "03e36c7f30aa"
+down_revision = "0242ac120002"
+branch_labels = None
+depends_on = None
+airflow_version = "3.1.0"
+
+
+def upgrade():
+    """Add is_active column to dag_version and backfill to True."""
+    op.add_column(
+        "dag_version",
+        sa.Column(
+            "is_active",
+            sa.Boolean(),
+            nullable=False,
+            server_default="1",
+            comment="Soft-delete flag; only active versions show up in APIs",
+        ),
+    )
+    # Backfill existing rows to active
+    op.execute("UPDATE dag_version SET is_active = 1")
+
+
+def downgrade():
+    """Remove is_active column from dag_version."""
+    op.drop_column("dag_version", "is_active")

airflow-core/src/airflow/migrations/versions/0071_3_1_0_add_is_active_to_dag_version.py

@@ -0,0 +1,72 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Modify task_instance.dag_version_id FK ondelete behavior to SET NULL.
+
+Revision ID: 1b2cd3e4f5a6
+Revises: 03e36c7f30aa
+Create Date: 2025-05-21 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from alembic import op
+from sqlalchemy_utils import UUIDType
+
+# revision identifiers, used by Alembic.
+revision = "1b2cd3e4f5a6"
+down_revision = "03e36c7f30aa"
+branch_labels = None
+depends_on = None
+airflow_version = "3.1.0"
+
+
+def upgrade():
+    """Alter task_instance.dag_version_id to SET NULL on delete."""
+    with op.batch_alter_table("task_instance", schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f("task_instance_dag_version_id_fkey"), type_="foreignkey")
+        batch_op.alter_column(
+            "dag_version_id",
+            existing_type=UUIDType(binary=False),
+            nullable=True,
+        )
+        batch_op.create_foreign_key(
+            batch_op.f("task_instance_dag_version_id_fkey"),
+            "dag_version",
+            ["dag_version_id"],
+            ["id"],
+            ondelete="SET NULL",
+        )
+
+
+def downgrade():
+    """Revert task_instance.dag_version_id FK ondelete back to CASCADE."""
+    with op.batch_alter_table("task_instance", schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f("task_instance_dag_version_id_fkey"), type_="foreignkey")
+        batch_op.alter_column(
+            "dag_version_id",
+            existing_type=UUIDType(binary=False),
+            nullable=False,
+        )
+        batch_op.create_foreign_key(
+            batch_op.f("task_instance_dag_version_id_fkey"),
+            "dag_version",
+            ["dag_version_id"],
+            ["id"],
+            ondelete="CASCADE",
+        )

airflow-core/src/airflow/migrations/versions/0072_3_1_0_modify_taskinstance_dag_version_fk.py

@@ -1973,7 +1973,10 @@ class DagModel(Base):
         "scheduler", "max_dagruns_to_create_per_loop", fallback=10
     )
     dag_versions = relationship(
-        "DagVersion", back_populates="dag_model", cascade="all, delete, delete-orphan"
+        "DagVersion",
+        back_populates="dag_model",
+        cascade="save-update, merge",
+        passive_deletes=True,
     )
 
     def __init__(self, **kwargs):

airflow-core/src/airflow/models/dag.py

@@ -21,7 +21,7 @@
 from typing import TYPE_CHECKING
 
 import uuid6
-from sqlalchemy import Column, ForeignKey, Integer, UniqueConstraint, select
+from sqlalchemy import Boolean, Column, ForeignKey, Integer, UniqueConstraint, select
 from sqlalchemy.orm import joinedload, relationship
 from sqlalchemy_utils import UUIDType
 
@@ -68,6 +68,13 @@ class DagVersion(Base):
     __table_args__ = (
         UniqueConstraint("dag_id", "version_number", name="dag_id_v_name_v_number_unique_constraint"),
     )
+    is_active = Column(
+        Boolean,
+        nullable=False,
+        default=True,
+        server_default="1",
+        comment="Soft-delete flag; only active versions show up in APIs",
+    )
 
     def __repr__(self):
         """Represent the object as a string."""
@@ -121,7 +128,7 @@ def _latest_version_select(
         :param dag_id: The DAG ID.
         :return: The select object.
         """
-        query = select(cls).where(cls.dag_id == dag_id)
+        query = select(cls).where(cls.dag_id == dag_id, cls.is_active)
         if bundle_version:
             query = query.where(cls.bundle_version == bundle_version)
 
@@ -163,16 +170,18 @@ def get_version(
         session: Session = NEW_SESSION,
     ) -> DagVersion | None:
         """
-        Get the version of the DAG.
+        Get the version of the DAG, if version_number is given, otherwise only active versions.
 
         :param dag_id: The DAG ID.
         :param version_number: The version number.
         :param session: The database session.
         :return: The version of the DAG or None if not found.
         """
         version_select_obj = select(cls).where(cls.dag_id == dag_id)
-        if version_number:
+        if version_number is not None:
             version_select_obj = version_select_obj.where(cls.version_number == version_number)
+        else:
+            version_select_obj = version_select_obj.where(cls.is_active)
 
         return session.scalar(version_select_obj.order_by(cls.id.desc()).limit(1))
 

airflow-core/src/airflow/models/dag_version.py

@@ -639,7 +639,11 @@ class TaskInstance(Base, LoggingMixin):
     next_kwargs = Column(MutableDict.as_mutable(ExtendedJSON))
 
     _task_display_property_value = Column("task_display_name", String(2000), nullable=True)
-    dag_version_id = Column(UUIDType(binary=False), ForeignKey("dag_version.id", ondelete="CASCADE"))
+    dag_version_id = Column(
+        UUIDType(binary=False),
+        ForeignKey("dag_version.id", ondelete="SET NULL"),
+        nullable=True,
+    )
     dag_version = relationship("DagVersion", back_populates="task_instances")
 
     __table_args__ = (

airflow-core/src/airflow/models/taskinstance.py

@@ -93,7 +93,7 @@ class MappedClassProtocol(Protocol):
     "2.10.0": "22ed7efa9da2",
     "2.10.3": "5f2621c13b39",
     "3.0.0": "29ce7909c52b",
-    "3.1.0": "0242ac120002",
+    "3.1.0": "1b2cd3e4f5a6",
 }