[takayoshi-makabe]

This prevents a Dag parsed from one file from overwriting an existing active Dag with the same dag_id from a different file.

When the Dag processor parses files one at a time, duplicate dag_id checks inside a single DagBag can miss duplicates that live in separate files. Before this change, syncing the later parsed file to the metadata database could update the existing DagModel and serialized Dag row, effectively making the last parsed file win.

This change adds a database-level collision check during Dag parsing result sync. If an active Dag with the same dag_id already exists for a different file, the incoming Dag is skipped and recorded as an import error using the existing AirflowDagDuplicatedIdException behavior.

related: #29321

Tests

Added a regression test covering this flow:

• sync an original Dag from original.py
• sync another Dag with the same dag_id from duplicate.py
• verify the original DagModel and serialized Dag remain unchanged
• verify the duplicate file gets a persisted import error

Validation run locally:

• uv run --project airflow-core ruff format airflow-core/src/airflow/dag_processing/collection.py airflow-core/tests/unit/dag_processing/test_collection.py
• uv run --project airflow-core ruff check --fix airflow-core/src/airflow/dag_processing/collection.py airflow-core/tests/unit/dag_processing/test_collection.py
• breeze run pytest airflow-core/tests/unit/dag_processing/test_collection.py::TestUpdateDagParsingResults::test_duplicate_dag_id_from_different_file_is_import_error -xvs

---

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

• Yes — Codex (GPT-5), used as an implementation assistant for code changes.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[uranusjr]

How would this work if you move a dag to a different file? To make this situation not cause an import error, the dag processor must remove the previous entry, but how can the processor know the previous dag is gone before parsing the file?

[yuseok89]

Suggested change

	"""Find Dag IDs already used by a different active DAG file."""
	"""Find Dag IDs already used by a different active Dag file."""

nit: "DAG file" -> "Dag file" to match the title-case convention used elsewhere

[takayoshi-makabe]

I fixed ea7b5bd

[takayoshi-makabe]

@uranusjr

Thanks for the question. The concrete scenario I had in mind is:

before: file_a.py → dag_id="hoge"

after: file_a.py → dag_id="fuga" (hoge removed, fuga added)
file_b.py → dag_id="hoge" (hoge moved here)

If file_b.py is parsed before file_a.py, this PR raises a transient import error on file_b.py since hoge → file_a.py is still active in the DB. However, this self-heals automatically: once file_a.py is re-parsed, hoge's last_parsed_time stops being updated, and DagProcessorManager._scan_stale_dags marks it stale after stale_dag_threshold seconds — so the next parse of file_b.py succeeds without collision.

So the current implementation is already eventually consistent in periodic-parsing environments.

What I'd rather avoid is silently letting the new file overwrite the existing entry — that's what current Airflow does today, and it's exactly how our team ended up with a broken DAG that nobody noticed. Making the collision visible as an import error, even transiently, is the whole point of this PR.

If the transient error window is a concern, one option would be to proactively mark disappeared DAGs as stale inside update_dag_parsing_results_in_db at parse time, which would reduce the window to just the next parse of the original file.