Migration fails when same dag id appears in two files

[dstandish]

From @vatsrahul1001

Migration from 2.10.5 fails.

The issue occurs when the same DAG exists in two different files. In such cases, the condition dag_code.fileloc_hash = serialized_dag.fileloc_hash fails. As a result, the dag_id column in the dag_code table remains null for that specific DAG, leading to the following error:

sqlalchemy.exc.IntegrityError: (psycopg2.errors.NotNullViolation) column "dag_id" of relation "dag_code" contains null values.

Traceback

[2025-04-19T15:27:59.511+0000] {migration.py:207} INFO - Context impl PostgresqlImpl.
[2025-04-19T15:27:59.511+0000] {migration.py:210} INFO - Will assume transactional DDL.
[2025-04-19T15:27:59.529+0000] {migration.py:618} INFO - Running upgrade d03e4a635aa3 -> 2b47dc6bc8df, add dag versioning.
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.NotNullViolation: column "dag_id" of relation "dag_code" contains null values


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 10, in <module>
    sys.exit(main())
             ^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/__main__.py", line 55, in main
    args.func(args)
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/cli_config.py", line 48, in command
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/cli.py", line 111, in wrapper
    return f(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/providers_configuration_loader.py", line 55, in wrapped_function
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/commands/db_command.py", line 197, in migratedb
    run_db_migrate_command(args, db.upgradedb, _REVISION_HEADS_MAP)
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/commands/db_command.py", line 125, in run_db_migrate_command
    command(
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/session.py", line 101, in wrapper
    return func(*args, session=session, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/db.py", line 1142, in upgradedb
    command.upgrade(config, revision=to_revision or "heads")
  File "/usr/local/lib/python3.12/site-packages/alembic/command.py", line 408, in upgrade
    script.run_env()
  File "/usr/local/lib/python3.12/site-packages/alembic/script/base.py", line 586, in run_env
    util.load_python_file(self.dir, "env.py")
  File "/usr/local/lib/python3.12/site-packages/alembic/util/pyfiles.py", line 95, in load_python_file
    module = load_module_py(module_id, path)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/alembic/util/pyfiles.py", line 113, in load_module_py
    spec.loader.exec_module(module)  # type: ignore
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<frozen importlib._bootstrap_external>", line 999, in exec_module
  File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
  File "/usr/local/lib/python3.12/site-packages/airflow/migrations/env.py", line 138, in <module>
    run_migrations_online()
  File "/usr/local/lib/python3.12/site-packages/airflow/migrations/env.py", line 132, in run_migrations_online
    context.run_migrations()
  File "<string>", line 8, in run_migrations
  File "/usr/local/lib/python3.12/site-packages/alembic/runtime/environment.py", line 946, in run_migrations
    self.get_context().run_migrations(**kw)
  File "/usr/local/lib/python3.12/site-packages/alembic/runtime/migration.py", line 623, in run_migrations
    step.migration_fn(**kw)
  File "/usr/local/lib/python3.12/site-packages/airflow/migrations/versions/0047_3_0_0_add_dag_versioning.py", line 238, in upgrade
    with op.batch_alter_table("dag_code") as batch_op:
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 144, in __exit__
    next(self.gen)
  File "/usr/local/lib/python3.12/site-packages/alembic/operations/base.py", line 397, in batch_alter_table
    impl.flush()
  File "/usr/local/lib/python3.12/site-packages/alembic/operations/batch.py", line 115, in flush
    fn(*arg, **kw)
  File "/usr/local/lib/python3.12/site-packages/alembic/ddl/postgresql.py", line 188, in alter_column
    super().alter_column(
  File "/usr/local/lib/python3.12/site-packages/alembic/ddl/impl.py", line 280, in alter_column
    self._exec(
  File "/usr/local/lib/python3.12/site-packages/alembic/ddl/impl.py", line 246, in _exec
    return conn.execute(construct, params)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/future/engine.py", line 286, in execute
    return self._execute_20(
           ^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1710, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/sql/ddl.py", line 80, in _execute_on_connection
    return connection._execute_ddl(
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1477, in _execute_ddl
    ret = self._execute_context(
          ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1953, in _execute_context
    self._handle_dbapi_exception(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 2134, in _handle_dbapi_exception
    util.raise_(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.NotNullViolation) column "dag_id" of relation "dag_code" contains null values

[SQL: ALTER TABLE dag_code ALTER COLUMN dag_id SET NOT NULL]
(Background on this error at: https://sqlalche.me/e/14/gkpj)
Stream closed EOF for absolute-pulsar-3831/lute-pulsar-3831-run-airflow-migrations3-0-1-alpha10-cjdqd (run-airflow-migrations)

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[dstandish]

side problem:
if migration fails, and you try to do it again, you'll get this error

[SQL:
CREATE TABLE _alembic_tmp_dag_code (
	fileloc_hash BIGINT NOT NULL,
	fileloc VARCHAR(2000) NOT NULL,
	last_updated TIMESTAMP NOT NULL,
	source_code TEXT NOT NULL,
	id CHAR(32),
	dag_version_id CHAR(32),
	source_code_hash VARCHAR(32),
	dag_id VARCHAR(250),
	created_at TIMESTAMP
)

]
(Background on this error at: https://sqlalche.me/e/14/e3q8)```

[dstandish]

The issue is that we still create a dag code record even when there's no serdag from it

dagcode

serdag

dag

Notice here that the dag even points to a diff file from the serdag