Fix dag-processor crash when renaming DAG tag case on MySQL

When a user changed only the case of a DAG tag (e.g., 'dangerous' to
'DANGEROUS'), the dag-processor would crash with a duplicate key error
on MySQL due to case-insensitive collation in the PRIMARY KEY. This
occurred because SQLAlchemy executed INSERT operations before DELETE
operations during the flush.

The fix ensures DELETE operations complete before attempting INSERT
operations by explicitly flushing and refreshing the tag relationship
from the database.

Fixes #56940

Author: kaxil

airflow-core/src/airflow/dag_processing/collection.py

@@ -162,7 +162,15 @@ def _update_dag_tags(tag_names: set[str], dm: DagModel, *, session: Session) ->
     for name, orm_tag in orm_tags.items():
         if name not in tag_names:
             session.delete(orm_tag)
-    dm.tags.extend(DagTag(name=name, dag_id=dm.dag_id) for name in tag_names.difference(orm_tags))
+            # Force DELETE operations to execute before INSERT operations.
+            # This prevents duplicate key errors on MySQL when renaming tags with only case changes,
+            # since MySQL uses case-insensitive collation for the (name, dag_id) primary key by default.
+            session.flush()
+            # Refresh the tags relationship from the database to reflect the deletions.
+            # Without this, dm.tags would still contain deleted tags in memory.
+            session.expire(dm, ["tags"])
+    current_tag_names = {t.name for t in dm.tags}
+    dm.tags.extend(DagTag(name=name, dag_id=dm.dag_id) for name in tag_names.difference(current_tag_names))
 
 
 def _update_dag_owner_links(dag_owner_links: dict[str, str], dm: DagModel, *, session: Session) -> None:

airflow-core/tests/unit/dag_processing/test_collection.py

@@ -37,6 +37,7 @@
     AssetModelOperation,
     DagModelOperation,
     _get_latest_runs_stmt,
+    _update_dag_tags,
     update_dag_parsing_results_in_db,
 )
 from airflow.exceptions import SerializationError
@@ -48,6 +49,7 @@
     DagScheduleAssetNameReference,
     DagScheduleAssetUriReference,
 )
+from airflow.models.dag import DagTag
 from airflow.models.errors import ParseImportError
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.providers.standard.operators.empty import EmptyOperator
@@ -941,3 +943,33 @@ def test_max_consecutive_failed_dag_runs_defaults_from_conf_when_none(
             update_dag_parsing_results_in_db("testing", None, [dag], {}, 0.1, set(), session)
             orm_dag = session.get(DagModel, "dag_max_failed_runs_default")
             assert orm_dag.max_consecutive_failed_dag_runs == 6
+
+
+@pytest.mark.db_test
+class TestUpdateDagTags:
+    @pytest.fixture(autouse=True)
+    def setup_teardown(self, session):
+        yield
+        session.query(DagModel).filter(DagModel.dag_id == "test_dag").delete()
+        session.commit()
+
+    @pytest.mark.parametrize(
+        ["initial_tags", "new_tags", "expected_tags"],
+        [
+            (["dangerous"], {"DANGEROUS"}, {"DANGEROUS"}),
+            (["existing"], {"existing", "new"}, {"existing", "new"}),
+            (["tag1", "tag2"], {"tag1"}, {"tag1"}),
+            (["keep", "remove", "lowercase"], {"keep", "LOWERCASE", "new"}, {"keep", "LOWERCASE", "new"}),
+            (["tag1", "tag2"], set(), set()),
+        ],
+    )
+    def test_update_dag_tags(self, testing_dag_bundle, session, initial_tags, new_tags, expected_tags):
+        dag_model = DagModel(dag_id="test_dag", bundle_name="testing")
+        dag_model.tags = [DagTag(name=tag, dag_id="test_dag") for tag in initial_tags]
+        session.add(dag_model)
+        session.commit()
+
+        _update_dag_tags(new_tags, dag_model, session=session)
+        session.commit()
+
+        assert {t.name for t in dag_model.tags} == expected_tags