Fix dag_tag unique-constraint crash when multiple dag processors parse concurrently

[kjh0623]

With multiple dag processors running in HA (e.g. replicas: 2), two processors can parse the same file concurrently: both read the (empty or stale) DagModel.tags relationship, then both INSERT the same dag_tag row. The plain ORM INSERT raises a unique-constraint violation (psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "dag_tag_pkey"), which poisons the session, surfaces as PendingRollbackError on the retry, and crashes the dag processor.

We hit this in production while running two dag processors over ~2,900 DAGs and have been running this fix as an internal patch since; this upstreams it.

Fix

• _update_dag_tags now inserts tags with a dialect-aware conflict-ignoring statement — ON CONFLICT DO NOTHING on PostgreSQL/SQLite, self-assign ON DUPLICATE KEY UPDATE on MySQL — following the existing pattern used for AssetActive in the same module (activate_assets_if_possible).
• The insert is batched across all dags in update_dags (the helper now takes the whole batch), so bulk_write_to_db keeps its constant query count — the assert_queries_count guards in test_dag.py stay at their current values.
• Pending tag DELETEs are flushed before the INSERT. This also guarantees deletes hit the database before inserts for case-only renames on MySQL (case-insensitive (name, dag_id) primary key), so the MySQL special-case block from Fix dag-processor crash when renaming DAG tag case on MySQL #57113 is folded into the new flow, and it ensures not-yet-flushed DagModel rows exist to satisfy the dag_tag FK.
• The relationship is expired after the insert; bulk_write_to_db already re-fetches the dags with joinedload(DagModel.tags) right after update_dags, so this adds no extra queries.

Tests

• Regression test simulating the HA race deterministically: the tags relationship is loaded, then the same tag is inserted behind the ORM's back before _update_dag_tags runs — previously IntegrityError, now tolerated with the remaining tags still applied.
• Test that tag changes for several dags are applied in one batched call.
• Test that a not-yet-flushed DagModel is flushed before the tag insert (FK safety).
• The existing parametrized test_update_dag_tags cases (including MySQL case-only renames) cover the folded-in Fix dag-processor crash when renaming DAG tag case on MySQL #57113 behaviour across all backend matrices.

closes: #68263
related: #50815

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

@kjh0623 This PR has been converted to draft because it does not yet meet our Pull Request quality criteria.

Issues found:

• ❌ Pre-commit / static checks: Failing: CI image checks / Static checks. Run prek run --from-ref main --stage pre-commit locally to reproduce and fix.
• ❌ Unit tests: Failing: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization, Postgres tests: core / DB-core:Postgres:14:3.10:Core...Serialization. Run breeze run pytest <path> -xvs locally.

Note: Your branch is 61 commits behind main. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.

What to do next:

• Fix each issue listed above.
• Make sure static checks pass locally (prek run --from-ref main --stage pre-commit).
• Mark the PR as "Ready for review" when you're done.

Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.