Alter task_instance.dag_version_id FK to default on delete.

Author: sunank200

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-7d4a160f7d5f4b017496e942255797d8e6b70eeb0d7c861739a66d736f549799
+a4f4017347264b506d701f647142ce96f7a1962aacc32d73a70ee4bdc6477e70

airflow-core/docs/migrations-ref.rst

@@ -40,7 +40,7 @@ Here's the list of all the Database Migrations that are executed via when you ru
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
 | ``1b2cd3e4f5a6`` (head) | ``03e36c7f30aa`` | ``3.1.0``         | Modify task_instance.dag_version_id FK on delete behavior to |
-|                         |                  |                   | SET NULL.                                                    |
+|                         |                  |                   | default (no SET NULL).                                       |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``03e36c7f30aa``        | ``0242ac120002`` | ``3.1.0``         | Add soft-delete flag to DagVersion (is_active).              |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

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

@@ -191,8 +191,6 @@ 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/migrations/versions/0072_3_1_0_modify_taskinstance_dag_version_fk.py

@@ -16,7 +16,7 @@
 # under the License.
 
 """
-Modify task_instance.dag_version_id FK on delete behavior to SET NULL.
+Modify task_instance.dag_version_id FK on delete behavior to default (no SET NULL).
 
 Revision ID: 1b2cd3e4f5a6
 Revises: 03e36c7f30aa
@@ -37,7 +37,7 @@
 
 
 def upgrade():
-    """Alter task_instance.dag_version_id to SET NULL on delete."""
+    """Alter task_instance.dag_version_id FK to default 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(
@@ -50,7 +50,6 @@ def upgrade():
             "dag_version",
             ["dag_version_id"],
             ["id"],
-            ondelete="SET NULL",
         )
 
 

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

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