From 8ed94069f2fc4072d7691b635204c80e0fd246ae Mon Sep 17 00:00:00 2001 From: Aaryan Mahajan Date: Wed, 22 Jul 2026 16:31:50 +0400 Subject: [PATCH 1/5] Fix db migrate failure to Airflow 2.9.2 under PyMySQL driver Upgrading through revision 686269002441 raised pymysql.err.ProgrammingError because the migration used MySQL's PREPARE/EXECUTE/DEALLOCATE PREPARE dynamic SQL to conditionally drop unique constraints, which PyMySQL's driver does not support (see https://github.com/PyMySQL/PyMySQL/issues/202). Replaced with a plain lookup of existing constraints from information_schema.TABLE_CONSTRAINTS followed by conditional ALTER TABLE ... DROP INDEX statements, which works with any MySQL driver. Verified against a real MySQL/PyMySQL backend: downgrading to bff083ad727d and re-running db migrate now completes without error. closes: #43690 --- ...istency_between_ORM_and_migration_files.py | 121 ++++++------------ 1 file changed, 37 insertions(+), 84 deletions(-) diff --git a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py index 64531a957b302..7197ca41188fe 100644 --- a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py +++ b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py @@ -44,35 +44,23 @@ def upgrade(): conn = op.get_bind() if conn.dialect.name == "mysql": # TODO: Rewrite these queries to use alembic when lowest MYSQL version supports IF EXISTS - conn.execute( - sa.text(""" - set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS WHERE - CONSTRAINT_SCHEMA = DATABASE() AND - TABLE_NAME = 'connection' AND - CONSTRAINT_NAME = 'unique_conn_id' AND - CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE connection - DROP INDEX unique_conn_id','select 1'); - - prepare stmt from @var; - execute stmt; - deallocate prepare stmt; - """) - ) - # Dropping the below and recreating cause there's no IF NOT EXISTS in mysql - conn.execute( - sa.text(""" - set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS WHERE - CONSTRAINT_SCHEMA = DATABASE() AND - TABLE_NAME = 'connection' AND - CONSTRAINT_NAME = 'connection_conn_id_uq' AND - CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE connection - DROP INDEX connection_conn_id_uq','select 1'); - - prepare stmt from @var; - execute stmt; - deallocate prepare stmt; + # `prepare`/`execute`/`deallocate prepare` is not supported by PyMySQL, so look up the + # existing unique constraints first and only drop the ones that are actually present. + existing_indexes = { + row[0] + for row in conn.execute( + sa.text(""" + SELECT CONSTRAINT_NAME FROM information_schema.TABLE_CONSTRAINTS + WHERE CONSTRAINT_SCHEMA = DATABASE() + AND TABLE_NAME = 'connection' + AND CONSTRAINT_TYPE = 'UNIQUE' """) - ) + ) + } + # Dropping the below and recreating cause there's no IF NOT EXISTS in mysql + for index_name in ("unique_conn_id", "connection_conn_id_uq"): + if index_name in existing_indexes: + conn.execute(sa.text(f"ALTER TABLE connection DROP INDEX {index_name}")) elif conn.dialect.name == "sqlite": # SQLite does not support DROP CONSTRAINT # We have to recreate the table without the constraint @@ -121,63 +109,28 @@ def upgrade(): batch_op.drop_constraint("task_reschedule_dr_fkey", type_="foreignkey") if conn.dialect.name == "mysql": - conn.execute( - sa.text(""" - set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS WHERE - CONSTRAINT_SCHEMA = DATABASE() AND - TABLE_NAME = 'dag_run' AND - CONSTRAINT_NAME = 'dag_run_dag_id_execution_date_uq' AND - CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE dag_run - DROP INDEX dag_run_dag_id_execution_date_uq','select 1'); - - prepare stmt from @var; - execute stmt; - deallocate prepare stmt; - """) - ) - conn.execute( - sa.text(""" - set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS WHERE - CONSTRAINT_SCHEMA = DATABASE() AND - TABLE_NAME = 'dag_run' AND - CONSTRAINT_NAME = 'dag_run_dag_id_run_id_uq' AND - CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE dag_run - DROP INDEX dag_run_dag_id_run_id_uq','select 1'); - - prepare stmt from @var; - execute stmt; - deallocate prepare stmt; - """) - ) + # `prepare`/`execute`/`deallocate prepare` is not supported by PyMySQL, so look up the + # existing unique constraints first and only drop the ones that are actually present. + existing_dag_run_indexes = { + row[0] + for row in conn.execute( + sa.text(""" + SELECT CONSTRAINT_NAME FROM information_schema.TABLE_CONSTRAINTS + WHERE CONSTRAINT_SCHEMA = DATABASE() + AND TABLE_NAME = 'dag_run' + AND CONSTRAINT_TYPE = 'UNIQUE' + """) + ) + } # below we drop and recreate the constraints because there's no IF NOT EXISTS - conn.execute( - sa.text(""" - set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS WHERE - CONSTRAINT_SCHEMA = DATABASE() AND - TABLE_NAME = 'dag_run' AND - CONSTRAINT_NAME = 'dag_run_dag_id_execution_date_key' AND - CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE dag_run - DROP INDEX dag_run_dag_id_execution_date_key','select 1'); - - prepare stmt from @var; - execute stmt; - deallocate prepare stmt; - """) - ) - conn.execute( - sa.text(""" - set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS WHERE - CONSTRAINT_SCHEMA = DATABASE() AND - TABLE_NAME = 'dag_run' AND - CONSTRAINT_NAME = 'dag_run_dag_id_run_id_key' AND - CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE dag_run - DROP INDEX dag_run_dag_id_run_id_key','select 1'); - - prepare stmt from @var; - execute stmt; - deallocate prepare stmt; - """) - ) + for index_name in ( + "dag_run_dag_id_execution_date_uq", + "dag_run_dag_id_run_id_uq", + "dag_run_dag_id_execution_date_key", + "dag_run_dag_id_run_id_key", + ): + if index_name in existing_dag_run_indexes: + conn.execute(sa.text(f"ALTER TABLE dag_run DROP INDEX {index_name}")) with op.batch_alter_table("callback_request", schema=None) as batch_op: batch_op.alter_column( "processor_subdir", From 6560d1578637da8bea88dbcbeb09c3f0c00ace46 Mon Sep 17 00:00:00 2001 From: Aaryan Mahajan Date: Wed, 22 Jul 2026 16:41:17 +0400 Subject: [PATCH 2/5] Add newsfragment for MySQL PyMySQL migration fix --- airflow-core/newsfragments/70235.bugfix.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 airflow-core/newsfragments/70235.bugfix.rst diff --git a/airflow-core/newsfragments/70235.bugfix.rst b/airflow-core/newsfragments/70235.bugfix.rst new file mode 100644 index 0000000000000..1068b45545b6e --- /dev/null +++ b/airflow-core/newsfragments/70235.bugfix.rst @@ -0,0 +1 @@ +Fixed ``airflow db migrate`` failing with ``pymysql.err.ProgrammingError`` when migrating to Airflow 2.9.2 using the PyMySQL driver From 511a2019d045a4baf464911ec21b363423240cc0 Mon Sep 17 00:00:00 2001 From: Aaryan Mahajan Date: Wed, 22 Jul 2026 22:43:50 +0400 Subject: [PATCH 3/5] Fix offline migration for 2.9.2 ORM-consistency revision The PyMySQL online-mode fix in this migration queries information_schema.TABLE_CONSTRAINTS via conn.execute() to decide which legacy unique constraints to drop. In Alembic's offline (--sql) mode there is no live connection, so conn.execute() returns None and iterating over it raises TypeError, breaking the CI "Test offline migration" step. Guard the query-then-conditionally-drop logic behind context.is_offline_mode(), falling back to the original dynamic prepare/execute/deallocate SQL text for the offline branch, since that text is only ever emitted as script output there rather than executed directly. --- ...istency_between_ORM_and_migration_files.py | 77 +++++++++++-------- 1 file changed, 46 insertions(+), 31 deletions(-) diff --git a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py index 7197ca41188fe..4cd075b54ce34 100644 --- a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py +++ b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py @@ -28,7 +28,7 @@ from __future__ import annotations import sqlalchemy as sa -from alembic import op +from alembic import context, op from sqlalchemy import literal # revision identifiers, used by Alembic. @@ -39,28 +39,57 @@ airflow_version = "2.9.2" +def _mysql_drop_unique_constraint_if_exists(conn, table: str, index_name: str) -> None: + """ + Drop a MySQL unique constraint only if it is actually present. + + MySQL has no ``DROP INDEX IF EXISTS``, and PyMySQL does not support the + ``prepare``/``execute``/``deallocate prepare`` sequence in a single + ``cursor.execute()`` call, so the existence check and the drop are issued as two + separate single statements. In offline (``--sql``) mode there is no live connection + to query information_schema against, so the guarded dynamic SQL is emitted as literal + script text instead, to be run later through a real SQL client that supports + multi-statement scripts. + """ + if context.is_offline_mode(): + conn.execute( + sa.text(f""" + set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS WHERE + CONSTRAINT_SCHEMA = DATABASE() AND + TABLE_NAME = '{table}' AND + CONSTRAINT_NAME = '{index_name}' AND + CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE {table} + DROP INDEX {index_name}','select 1'); + + prepare stmt from @var; + execute stmt; + deallocate prepare stmt; + """) + ) + return + existing_indexes = { + row[0] + for row in conn.execute( + sa.text(f""" + SELECT CONSTRAINT_NAME FROM information_schema.TABLE_CONSTRAINTS + WHERE CONSTRAINT_SCHEMA = DATABASE() + AND TABLE_NAME = '{table}' + AND CONSTRAINT_TYPE = 'UNIQUE' + """) + ) + } + if index_name in existing_indexes: + conn.execute(sa.text(f"ALTER TABLE {table} DROP INDEX {index_name}")) + + def upgrade(): """Apply Update missing constraints.""" conn = op.get_bind() if conn.dialect.name == "mysql": # TODO: Rewrite these queries to use alembic when lowest MYSQL version supports IF EXISTS - # `prepare`/`execute`/`deallocate prepare` is not supported by PyMySQL, so look up the - # existing unique constraints first and only drop the ones that are actually present. - existing_indexes = { - row[0] - for row in conn.execute( - sa.text(""" - SELECT CONSTRAINT_NAME FROM information_schema.TABLE_CONSTRAINTS - WHERE CONSTRAINT_SCHEMA = DATABASE() - AND TABLE_NAME = 'connection' - AND CONSTRAINT_TYPE = 'UNIQUE' - """) - ) - } # Dropping the below and recreating cause there's no IF NOT EXISTS in mysql for index_name in ("unique_conn_id", "connection_conn_id_uq"): - if index_name in existing_indexes: - conn.execute(sa.text(f"ALTER TABLE connection DROP INDEX {index_name}")) + _mysql_drop_unique_constraint_if_exists(conn, "connection", index_name) elif conn.dialect.name == "sqlite": # SQLite does not support DROP CONSTRAINT # We have to recreate the table without the constraint @@ -109,19 +138,6 @@ def upgrade(): batch_op.drop_constraint("task_reschedule_dr_fkey", type_="foreignkey") if conn.dialect.name == "mysql": - # `prepare`/`execute`/`deallocate prepare` is not supported by PyMySQL, so look up the - # existing unique constraints first and only drop the ones that are actually present. - existing_dag_run_indexes = { - row[0] - for row in conn.execute( - sa.text(""" - SELECT CONSTRAINT_NAME FROM information_schema.TABLE_CONSTRAINTS - WHERE CONSTRAINT_SCHEMA = DATABASE() - AND TABLE_NAME = 'dag_run' - AND CONSTRAINT_TYPE = 'UNIQUE' - """) - ) - } # below we drop and recreate the constraints because there's no IF NOT EXISTS for index_name in ( "dag_run_dag_id_execution_date_uq", @@ -129,8 +145,7 @@ def upgrade(): "dag_run_dag_id_execution_date_key", "dag_run_dag_id_run_id_key", ): - if index_name in existing_dag_run_indexes: - conn.execute(sa.text(f"ALTER TABLE dag_run DROP INDEX {index_name}")) + _mysql_drop_unique_constraint_if_exists(conn, "dag_run", index_name) with op.batch_alter_table("callback_request", schema=None) as batch_op: batch_op.alter_column( "processor_subdir", From e792ca770884da38a014a2ac68b888f7d3f686b6 Mon Sep 17 00:00:00 2001 From: Aaryan Mahajan Date: Sun, 9 Aug 2026 18:53:10 +0530 Subject: [PATCH 4/5] Address review feedback on MySQL PyMySQL migration fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bind the table name as a query parameter in the information_schema lookup instead of interpolating it into the SQL string, since it's a value comparison rather than an identifier — matches Airflow's usual parameter-binding convention even though every call site currently passes a hardcoded literal. Also correct the docstring: PyMySQL can run a bare PREPARE statement fine, what it actually can't do is send more than one statement per cursor.execute() call, because it never enables CLIENT.MULTI_STATEMENTS. --- ..._inconsistency_between_ORM_and_migration_files.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py index 4cd075b54ce34..26d2e86176cf1 100644 --- a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py +++ b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py @@ -43,8 +43,9 @@ def _mysql_drop_unique_constraint_if_exists(conn, table: str, index_name: str) - """ Drop a MySQL unique constraint only if it is actually present. - MySQL has no ``DROP INDEX IF EXISTS``, and PyMySQL does not support the - ``prepare``/``execute``/``deallocate prepare`` sequence in a single + MySQL has no ``DROP INDEX IF EXISTS``, and PyMySQL does not enable + ``CLIENT.MULTI_STATEMENTS``, so it cannot run the + ``prepare``/``execute``/``deallocate prepare`` script in a single ``cursor.execute()`` call, so the existence check and the drop are issued as two separate single statements. In offline (``--sql``) mode there is no live connection to query information_schema against, so the guarded dynamic SQL is emitted as literal @@ -70,12 +71,13 @@ def _mysql_drop_unique_constraint_if_exists(conn, table: str, index_name: str) - existing_indexes = { row[0] for row in conn.execute( - sa.text(f""" + sa.text(""" SELECT CONSTRAINT_NAME FROM information_schema.TABLE_CONSTRAINTS WHERE CONSTRAINT_SCHEMA = DATABASE() - AND TABLE_NAME = '{table}' + AND TABLE_NAME = :table AND CONSTRAINT_TYPE = 'UNIQUE' - """) + """), + {"table": table}, ) } if index_name in existing_indexes: From 22271054271a8515571e9331fc3fa38adac8c743 Mon Sep 17 00:00:00 2001 From: Aaryan Mahajan Date: Thu, 20 Aug 2026 10:20:29 +0530 Subject: [PATCH 5/5] Fixed the PR comments --- airflow-core/newsfragments/70235.bugfix.rst | 1 - 1 file changed, 1 deletion(-) delete mode 100644 airflow-core/newsfragments/70235.bugfix.rst diff --git a/airflow-core/newsfragments/70235.bugfix.rst b/airflow-core/newsfragments/70235.bugfix.rst deleted file mode 100644 index 1068b45545b6e..0000000000000 --- a/airflow-core/newsfragments/70235.bugfix.rst +++ /dev/null @@ -1 +0,0 @@ -Fixed ``airflow db migrate`` failing with ``pymysql.err.ProgrammingError`` when migrating to Airflow 2.9.2 using the PyMySQL driver