Add cascade to dag_tag to dag foreignkey

airflow/migrations/utils.py

@@ -35,7 +35,8 @@ def get_mssql_table_constraints(conn, table_name):
      FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
      JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
      WHERE tc.TABLE_NAME = '{table_name}' AND
-     (tc.CONSTRAINT_TYPE = 'PRIMARY KEY' or UPPER(tc.CONSTRAINT_TYPE) = 'UNIQUE')
+     (tc.CONSTRAINT_TYPE = 'PRIMARY KEY' or UPPER(tc.CONSTRAINT_TYPE) = 'UNIQUE'
+     or UPPER(tc.CONSTRAINT_TYPE) = 'FOREIGN KEY')
     """
     result = conn.execute(query).fetchall()
     constraint_dict = defaultdict(lambda: defaultdict(list))

airflow/migrations/versions/0110_2_3_2_add_cascade_to_dag_tag_foreignkey.py

@@ -0,0 +1,84 @@
+#
+# 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 cascade to dag_tag foreign key
+
+Revision ID: 3c94c427fdf6
+Revises: 1de7bc13c950
+Create Date: 2022-05-03 09:47:41.957710
+
+"""
+
+from alembic import op
+
+from airflow.migrations.utils import get_mssql_table_constraints
+
+# revision identifiers, used by Alembic.
+revision = '3c94c427fdf6'
+down_revision = '1de7bc13c950'
+branch_labels = None
+depends_on = None
+airflow_version = '2.3.2'
+
+
+def upgrade():
+    """Apply Add cascade to dag_tag foreignkey"""
+    conn = op.get_bind()
+    if conn.dialect.name == 'sqlite':
+        naming_convention = {
+            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+        }
+        with op.batch_alter_table(
+            'dag_tag', naming_convention=naming_convention, recreate='always'
+        ) as batch_op:
+            batch_op.drop_constraint('fk_dag_tag_dag_id_dag', type_='foreignkey')
+            batch_op.create_foreign_key(
+                "dag_tag_dag_id_fkey", 'dag', ['dag_id'], ['dag_id'], ondelete='CASCADE'
+            )
+    else:
+        with op.batch_alter_table('dag_tag') as batch_op:
+            if conn.dialect.name == 'mssql':
+                constraints = get_mssql_table_constraints(conn, 'dag_tag')
+                Fk, _ = constraints['FOREIGN KEY'].popitem()
+                batch_op.drop_constraint(Fk, type_='foreignkey')
+            if conn.dialect.name == 'postgresql':
+                batch_op.drop_constraint('dag_tag_dag_id_fkey', type_='foreignkey')
+            if conn.dialect.name == 'mysql':
+                batch_op.drop_constraint('dag_tag_ibfk_1', type_='foreignkey')
+
+            batch_op.create_foreign_key(
+                "dag_tag_dag_id_fkey", 'dag', ['dag_id'], ['dag_id'], ondelete='CASCADE'
+            )
+
+
+def downgrade():
+    """Unapply Add cascade to dag_tag foreignkey"""
+    conn = op.get_bind()
+    if conn.dialect.name == 'sqlite':
+        with op.batch_alter_table('dag_tag') as batch_op:
+            batch_op.drop_constraint('dag_tag_dag_id_fkey', type_='foreignkey')
+            batch_op.create_foreign_key("fk_dag_tag_dag_id_dag", 'dag', ['dag_id'], ['dag_id'])
+    else:
+        with op.batch_alter_table('dag_tag') as batch_op:
+            batch_op.drop_constraint('dag_tag_dag_id_fkey', type_='foreignkey')
+            batch_op.create_foreign_key(
+                None,
+                'dag',
+                ['dag_id'],
+                ['dag_id'],
+            )

airflow/models/dag.py

@@ -2638,7 +2638,11 @@ class DagTag(Base):
 
     __tablename__ = "dag_tag"
     name = Column(String(100), primary_key=True)
-    dag_id = Column(String(ID_LEN), ForeignKey('dag.dag_id'), primary_key=True)
+    dag_id = Column(
+        String(ID_LEN),
+        ForeignKey('dag.dag_id', name='dag_tag_dag_id_fkey', ondelete='CASCADE'),
+        primary_key=True,
+    )
 
     def __repr__(self):
         return self.name
@@ -2689,7 +2693,7 @@ class DagModel(Base):
     timetable_description = Column(String(1000), nullable=True)
 
     # Tags for view filter
-    tags = relationship('DagTag', cascade='all,delete-orphan', backref=backref('dag'))
+    tags = relationship('DagTag', cascade='all, delete, delete-orphan', backref=backref('dag'))
 
     max_active_tasks = Column(Integer, nullable=False)
     max_active_runs = Column(Integer, nullable=True)

docs/apache-airflow/migrations-ref.rst

@@ -27,7 +27,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
 | Revision ID                     | Revises ID        | Airflow Version   | Description                                                  |
 +=================================+===================+===================+==============================================================+
-| ``1de7bc13c950`` (head)         | ``b1b348e02d07``  | ``2.3.1``         | Add index for ``event`` column in ``log`` table.             |
+| ``3c94c427fdf6`` (head)         | ``1de7bc13c950``  | ``2.3.2``         | Add cascade to dag_tag foreign key                           |
++---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
+| ``1de7bc13c950``                | ``b1b348e02d07``  | ``2.3.1``         | Add index for ``event`` column in ``log`` table.             |
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
 | ``b1b348e02d07``                | ``75d5ed6c2b43``  | ``2.3.0``         | Update dag.default_view to grid                              |
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+